数据流在本地返回正确的类型，但在云中执行时不返回

Posted 2023-03-25

【中文标题】数据流在本地返回正确的类型，但在云中执行时不返回

【英文标题】：Dataflow returns correct type locally, but not when executed in the cloud

【发布时间】：2015-11-26 11:34:04

【问题描述】：

鉴于 BigQuery 中的下表：

具有以下5个值：

还有一个简单的 ParDo 读取它并打印类型：

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.io.BigQueryIO;
import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.DataflowPipelineWorkerPoolOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.runners.BlockingDataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.transforms.DoFn;
import com.google.cloud.dataflow.sdk.transforms.ParDo;

public class FloatBug 
    public static void main(String[] args) 
        DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
        options.setRunner(BlockingDataflowPipelineRunner.class);
        options.setProject("<project_id>");
        options.setWorkerMachineType("n1-standard-1");
        options.setZone("us-central1-a");
        options.setStagingLocation("<gcs_bucket>");
        options.setNumWorkers(1);
        options.setMaxNumWorkers(1);
        options.setAutoscalingAlgorithm(DataflowPipelineWorkerPoolOptions.AutoscalingAlgorithmType.NONE);
        Pipeline pipeline = Pipeline.create(options);

        pipeline.apply(BigQueryIO.Read.from("FLOAT_BUG.float_bug")).apply(ParDo.of(new DoFn<TableRow, TableRow>() 
            @Override
            public void processElement(ProcessContext c) throws Exception 
                Object o = c.element().get("VHH");
                if (o instanceof Double) 
                    System.out.println("Awesome. Got expected Double: " + o);
                 else if (o instanceof Integer) 
                    System.out.println("Bummer. Got an Integer: " + o);
                 else 
                    assert false;
                
            
        ));
        pipeline.run();
    

在本地运行会为 每个 值返回一个 Double。这就是我所期望的：

Awesome. Got expected Double: 2.0
Awesome. Got expected Double: 2.245
Awesome. Got expected Double: 1.773
Awesome. Got expected Double: 4.567
Awesome. Got expected Double: 1.342

但是，使用 Dataflow 服务在 云 中运行会返回 Integer 值 2.0：

Awesome. Got expected Double: 2.245
Awesome. Got expected Double: 1.342
Awesome. Got expected Double: 1.773
Awesome. Got expected Double: 4.567
Bummer. Got an Integer: 2

它应该返回 Double，而不是 2.0 的 Integer

【参考方案1】：

观察结果是正确的。从 BigQuery 读取输入的管道可能会输出与 BigQuery 架构中的基础数据类型不同类型的数据。正如观察到的那样，类型也可能因元素而异。

这是一个不幸的结果，因为 Dataflow Service 首先将数据从 BigQuery 导出到 Google Cloud Storage 中的 JSON 编码文件，然后从这些文件中读取数据。 JSON 显然不保留类型。例如，浮点数2.0 将被编码为字符串"2"，在Java 中将被读取为Integer。使用 DirectPipelineRunner 执行管道时不会发生这种情况，因为该运行程序直接从 BigQuery 读取。

现在，避免此类问题的最简单方法是通过 Java 中的 Number 抽象类。这是Double 和Integer 等类的超类。将结果解释为Number 然后调用doubleValue() 方法应该是安全的。

---

也就是说，今后，我希望这种行为会有所改变。确切的时间线尚不清楚，但数据流服务的行为应该会很快与本地执行相匹配。通过Number 类的解决方法应该是正确的。

【讨论】：

谢谢达沃尔。是的，我们已经开始将所有代码更改为使用 Number。但我们真的不应该这样做。它应该在不同的跑步者之间保持一致，即本地和云端

我完全同意。您将在此处了解内部实现细节。我们会修复它，只是还不能保证时间表。