如何在数据流中正确使用“展平" [英] How to use 'flatten' correctly in Dataflow

问题描述

我们的管道如下所示:

GCS(gz压缩文件)-> ParDo-> BigQuery

我想使用'flatten'从GCS抓取多个文件作为管道的输入.但它总是让错误感到厌烦:

I want to use a 'flatten' to scoop in multiple files from GCS as the input to my pipeline. But it keeps borking with the error:

Workflow failed. Causes: (5001e5764f46ac2c): BigQuery creation of import job for table "Impressions_05_2015_denormalized_test" in dataset "CPT_XXXX" in project "gdfp-XXXX" failed. Causes: (5001e5764f46a1cf): Error:
 Message: Load configuration must specify at least one source URI
 HTTP Code: 400

代码:

PCollection<String> file1 = pipeline.apply(TextIO.Read.from("gs://<bucket_name_removed>/NetworkActiveViews_232503_20140918_21.gz").withCompressionType(TextIO.CompressionType.GZIP));
        PCollection<String> file2 = pipeline.apply(TextIO.Read.from("gs://<bucket_name_removed>/NetworkActiveViews_232503_20140918_22.gz").withCompressionType(TextIO.CompressionType.GZIP));
        PCollectionList<String> allFiles = PCollectionList.of(file1).and(file2);
        PCollection<String> inputRead = allFiles.apply(Flatten.<String>pCollections());
inputRead.apply(ParDo.of(transformation)
                .named(String.format("%s-CPT-transform", type))
                .withSideInputs(views))
                .apply(Write.to(getOutputTable(type))
                        .withCreateDisposition(CREATE_IF_NEEDED)
                        .withWriteDisposition(WRITE_APPEND)
                        .withSchema(schema)
                        .named(String.format("%s-BQ-write", type)));

示例工作ID: 2015-05-12_19_54_06-10158770219525037626

我在做什么错了?

推荐答案

我提出的hack并不是提议的hack，这确实很粗糙，而是在finishBundle()方法中写了一个空行.这将在每个捆绑软件中写入1个空行，但是我们可以忍受，直到推出修复程序为止.设置"id"可以使以后更轻松地过滤掉这些行.

Instead of the hack proposed, which is really quite gnarly, I instead write a single empty row in the finishBundle() method. This will write 1 empty row per bundle, but we can live with that until the fix is rolled out. Setting the "id" makes it much easier to filter those rows out later on.

此外，这种解决方法/黑客更容易实现:

Also, this workaround/hack is much easier to implement:

@Override
public void finishBundle(Context c) throws Exception {
   TableRow workaroundRow = new TableRow();
   workaroundRow.set("id", "workaround_row");
   c.output(workaroundRow); //Workaround to http://goo.gl/CpBxEf
}

这篇关于如何在数据流中正确使用“展平"的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！