Google Dataflow - 在当前运行的步骤中并行化工作的能力

Question

我正在尝试为模拟过程运行 google 数据流。我尝试通过最大工作节点 5 启动数据流，并启用自动缩放（THROUGHPUT_BASED）。

问题是数据流没有利用所有工作节点 (5) 广告给我消息。

自动缩放：根据在当前运行的步骤中并行化工作的能力，将工作人员的数量减少到 1 个。

请提出可能是什么问题。

日志附在下面。

2017-05-10T06:01:36.555Z：详细信息：(5de468ab73bd7581)：自动缩放是 已为作业 2017-05-09_23_01_36-6765647625807820060 启用。的数量 工人将在 1 到 1000 之间。 2017-05-10T06:01:37.811Z：详细信息： (feb335244b957ccc)：检查所需的云 API 是否已启用。 2017-05-10T06：02：05.328Z：详细信息：（feb335244b9573fd）：正在扩展 GroupByKey 操作成可优化的部分。 2017-05-10T06:02:05.332Z: 详细信息: (feb335244b957017): 起重 ValueCombiningMappingFns 到 MergeBucketsMappingFns 2017-05-10T06:02:05.714Z: 详细信息: (feb335244b957a72): 熔断相邻 ParDo、读取、写入和展平操作 2017-05-10T06:02:05.716Z： 详细信息：（feb335244b95768c）：将消费者 AnonymousParDo 融合到 读取/DataflowPipelineRunner.BatchBigQueryIONativeRead 2017-05-10T06:02:05.718Z: 详细信息: (feb335244b9572a6): 熔断消费者 WriteToBigQuery/DataflowPipelineRunner.BatchBigQueryIOWrite/DataflowPipelineRunner.BatchBigQueryIONativeWrite 进入 AnonymousParDo 2017-05-10T06:02:05.728Z：详细信息： (feb335244b95730e)：将 StepResource 设置和拆卸添加到工作流 图形。 2017-05-10T06：02：05.767Z：基本：（75edc98e2ecf8a50）：正在执行 手术 读取/DataflowPipelineRunner.BatchBigQueryIONativeRead+AnonymousParDo+WriteToBigQuery/DataflowPipelineRunner.BatchBigQueryIOWrite/DataflowPipelineRunner.BatchBigQueryIONativeWrite 2017-05-10T06：02：05.973Z：基本：（c9b9bc749d188107）：开始 5 工人...... 2017-05-10T06：02：07.735Z：基本：（4903a2f536f5c1ae）： BigQuery 查询作为作业发布：“dataflow_job_4354447435307355184”。你 可以使用 bq 工具检查其状态：“bq show -j --project_id=com-dl-parts-dev dataflow_job_4354447435307355184"。2017-05-10T06:03:03.727Z：详细信息：(376dc1d89cc4c16e)：工人有 成功启动。 2017-05-10T06：04：09.561Z：基本： (4903a2f536f5c868)：BigQuery 查询已完成，作业： “dataflow_job_4354447435307355184” 2017-05-10T06:04:09.987Z：基本： (4903a2f536f5cfbf)：BigQuery 导出作业 “dataflow_job_5261227963771635305”开始了。您可以查看其状态 使用 bq 工具：“bq show -j --project_id=com-dl-parts-dev dataflow_job_5261227963771635305"。2017-05-10T06:05:10.798Z：详细信息： (d0c3fbda1e314661)：BigQuery 导出作业进度： “dataflow_job_5261227963771635305”共观察到 1 个导出文件 迄今。 2017-05-10T06：05：10.801Z：基本：（d0c3fbda1e31498f）： BigQuery 导出作业已完成：“dataflow_job_5261227963771635305” 2017-05-10T06：06：12.303Z：基本：（d51372b306784d58）：自动缩放： 将工作池的大小从 5 调整为 1。 2017-05-10T06:06:12.306Z：详细信息： (d51372b306784f56)：自动缩放：将工作人员的数量减少到 1 基于当前运行步骤的进度。 2017-05-10T06：07：38.000Z：基本：（4e3c32839262218d）：自动缩放： 试图将工作人员池升级到 2。 2017-05-10T06:08:11.074Z: 基本：（4e3c328392622142）：自动缩放：将工作池的大小从 1 调整为 2. 2017-05-10T06:08:11.077Z: 详细信息: (4e3c32839262213c): Autoscaling: 根据并行化的能力将工作人员的数量提高到 2 当前正在运行的步骤中的工作。 2017-05-10T06:13:41.023Z: 基本：（ae82479176c127a3）：自动缩放：将工作池的大小从 2 调整为 1. 2017-05-10T06:13:41.027Z: 详情: (ae82479176c129d5): Autoscaling: 根据并行化的能力将工作人员的数量减少到 1 当前正在运行的步骤中的工作。 2017-05-10T06:20:31.844Z: 基本：（a41dfc71af8c36fd）：执行 BigQuery 导入作业 “dataflow_job_4354447435307352182”。您可以使用 bq 工具：“bq show -j --project_id=com-dl-parts-dev dataflow_job_4354447435307352182"。

更新：-这是防止融合的正确方法吗？

我正在使用 bigquery IO 读取 Bigquery 表。此输入参数为我提供每条记录中的产品编号。

在我进行 ParDo 操作之后。在流程元素函数中，我正在为从输入中获得的每个产品进行一些数据预测操作。

PCollection<TableRow> quotes3 = quotes2.apply(ParDo.of(new  DoFn<TableRow, TableRow>() {
private static final long serialVersionUID = 1L;
    @Override
    public void processElement(ProcessContext c) throws Exception{

        TableRow rowData = c.element();
        rowData = c.element();
        TableRow tableRowData = ForcastOperation(rowData);
        c.output(tableRowDRP);
    }
}));

在最后一步中，我使用数据流管道将预测结果转储到 bigquery 中。

quotes3.apply(BigQueryIO.Write
        .named("WriteToBigQuery")
        .to("com-dl-parts:ds_parts.output_data")
        .withSchema(schema)
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

pipelineTransform.run();

更新时间：- 17/05/2017 13:38

我正在尝试通过下面列出的方式打破融合。它将 GroupByKey 应用操作扩展到 308 个节点。但我不确定它是否将具有 ForcastOperation 方法的第二个 pardo 扩展到 308 个节点。

 PCollection<String> quotes1 = quotes.apply(ParDo.of(new  DoFn<TableRow, KV<String, String>>() {
    private static final long serialVersionUID = 1L;
    private Random random = new Random();

        @Override
        public void processElement(ProcessContext c) throws Exception{
            TableRow rowData = c.element();
            rowData = c.element();
            c.output(KV.of(rowData.get("APRODUCT").toString(), rowData.get("APRODUCT").toString()));
        }
        })).apply(GroupByKey.<String, String>create())
           .apply(Values.<Iterable<String>>create())
           .apply(Flatten.<String>iterables());



PCollection<TableRow> quotes3 = quotes1.apply(ParDo.of(new  DoFn<String, TableRow>() {
private static final long serialVersionUID = 1L;
    @Override
    public void processElement(ProcessContext c) throws Exception{

        TableRow rowData = c.element();
        rowData = c.element();
        TableRow tableRowData = ForcastOperation(rowData);
        c.output(tableRowDRP);
    }
}));

在最后一步中，我使用数据流管道将预测结果转储到 bigquery 中。

    quotes3.apply(BigQueryIO.Write
        .named("WriteToBigQuery")
        .to("com-dl-parts:ds_parts.output_data")
        .withSchema(schema)
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

pipelineTransform.run();

Answer 1

我查看了这项作业的日志，发现它从 BigQuery 作为输入读取的数据量非常小 - 大约 1kb。这是预期的吗？

如果是，并且如果您仍想并行处理 1kb 1000 路，那么我假设您的 ForecastOperation 函数每个元素的计算量都非常大。

在这种情况下，您需要打破从 BigQuery 读取数据和应用 ForecastOperation 之间的融合。请看https://cloud.google.com/dataflow/service/dataflow-service-desc#preventing-fusion