Apache束流管道中传递给BigQuery.IO连接器的动态参数

Question

嗨，这对于Apache来说是非常新的，它试图创建一个批处理管道，将增量数据从Big加载到扳手。

为此，我必须参考扳手来获取插入的最后一条记录的时间戳，并从该时间戳中从一个大查询中提取数据。

我正试图找到一种将参数传递给查询的方法，我在大型查询连接器中使用了这些参数，如下所示。

PCollection<Struct> metaInfo = getLastReadTimestamp(pipeline, options, uuid);
PCollection<TableRow> bqDataSet = getBQRecords(pipeline,options,metaInfo,uuid);
public PCollection<Struct> getLastReadTimestamp(Pipeline pipeline,
                                                CustomerMessageReaderOptions options,
                                                UUID uuid){

    PCollection<Struct> lastReadTimestamp = pipeline.apply("read from spanner",
            SpannerIO.read().withBatching(false)
                            .withInstanceId(options.getProject())
                            .withDatabaseId(options.getDataBaseName())
                            .withQuery("select max(create_datetime) as readtimestamp from customer_txn"));
    return lastReadTimestamp;
}

public PCollection<TableRow> getBQRecords(Pipeline pipeline,
                                          CustomerMessageReaderOptions options,
                                          PCollection<Struct> metadata,
                                          UUID uuid){
    String lastReadTime = "" ; //read from metadata
    String bqQuery = String.format("select * from `projectid.customerdataset.customer_txns where createtimestamp > '%s'`",lastReadTime);
    PCollection<TableRow> bqDataSet = pipeline.apply("read from BQ",
            BigQueryIO.readTableRows()
                        .withMethod(BigQueryIO.TypedRead.Method.DIRECT_READ)
                        .usingStandardSql()
                        .fromQuery(bqQuery));
    return bqDataSet;
}

因此，从上面的metaInfo( getBQRecords )代码中可以看到，我有一个时间戳，需要将它传递给查询，以便在getBQRecords方法中从Big读取数据。

我知道我不能像上面代码中所示的那样从pcollection中声明变量值，这是不可能的，但是我找到了解决这个问题的方法

我的问题是，我是否可以做到这一点，基本上，一种将值动态地传递到数据流作业本身中的查询参数的方法？

任何建议都会有帮助.！

提前谢谢

Answer 1

我提出了一个解决方案，希望能有所帮助。

在Kubernetes pod中，在使用Maven在命令行中启动Dataflow作业之前，可以执行以下步骤：

• 创建shell脚本
• 使用gcloud命令执行对Spanner的查询，以获得shell脚本中的最后一个日期时间：

F 213

gcloud spanner databases execute-sql your-database --instance=your-instance --sql='select max(create_datetime) as readtimestamp from customer_txn'

Spanner

• Pass将结果日期时间作为mvn命令行中的程序参数获得:

mvn compile exec:java \
  -Dexec.mainClass=package.MainClass \
  -Dexec.args=" \
  --project=project \
  --runner=DataflowRunner \
  --last_datetime=result_datetime_from_spanner
...

然后，在您的Beam. Beam Dataflow中，获取最后一个datetime，作为管道选项

• ，将此日期时间传递给SQL查询，以便在BigqueryIO中执行

此解决方案将Spanner日期时间检索的职责委托到Dataflow之外。它在启动部分增加了一些复杂性，但使Dataflow工作更容易。

除了拥有启动Dataflow作业的权限外，来自Kubernetes荚的执行器还必须具有从Spanner读取数据的权限。