在读取部分，可以将TableRow与JdbcIO一起使用，如下所示：

PCollection<TableRow> results = pipeline
        .apply("Connect", JdbcIO.<TableRow>read()
                .withDataSourceConfiguration(buildDataSourceConfig(options, URL))
                .withQuery(query)
                .withCoder(TableRowJsonCoder.of())
                .withRowMapper(new JdbcIO.RowMapper<TableRow>() {
                    @Override
                    public TableRow mapRow(ResultSet resultSet) {
                        // Implements your logic here.
                      
                        return tableRow;
                    }
                )
        );

我认为Dataflow不是满足您需求的更容易的服务，因为您必须向BigQueryIO传递一个模式，以允许Dataflow使用CREATE_IF_NEEDED选项创建表。
使用以下方法传递架构：withSchema(schema)，示例：

rows.apply(
        "Write to BigQuery",
        BigQueryIO.writeTableRows()
            .to(String.format("%s:%s.%s", project, dataset, table))
            .withSchema(schema)
            // For CreateDisposition:
            // - CREATE_IF_NEEDED (default): creates the table if it doesn't exist, a schema is
            // required
            // - CREATE_NEVER: raises an error if the table doesn't exist, a schema is not needed
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
            // For WriteDisposition:
            // - WRITE_EMPTY (default): raises an error if the table is not empty
            // - WRITE_APPEND: appends new rows to existing rows
            // - WRITE_TRUNCATE: deletes the existing rows before writing
            .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE));

    // pipeline.run().waitUntilFinish();
  }

这意味着在您的情况下，必须从PCollection中的当前TableRow构建模式。
使用Beam并不容易做到这一点，如果您可以为BigQuery表使用一个schema，那就更容易了。
我向你们提出另一种解决办法和途径：

• 将MS-SQL数据导出到Cloud Storage
• 使用autodetect模式将Cloud Storage文件导入BigQuery，以从文件推断架构。

如果希望使用gcloud cli自动执行此过程，可以在Shell脚本中添加此逻辑。
将GCS文件加载到BigQuery的示例：

bq load \
    --autodetect \
    --replace \
    --source_format=CSV \
    mydataset.mytable \
    gs://mybucket/mydata.csv