1
我們正在嘗試將我們的數據流/波束管線從2.0.0-beta3遷移到2.0.0。空側輸出在SDK 2.0.0上爲Dataflow/Beam投擲NPE
但是，當我們使用2.0.0版本時，管道會因數據流/ Beam API中深入的NPE而失敗。改回2.0.0-beta3，它再次正常工作。
對代碼進行的唯一更改是將2.0.0 SDK的API更改合併在一起。我們沒有改變任何其他事情。問題出現在側面輸出爲空時。空側輸出在2.0.0-beta3上正常工作。
我們在遷移到2.0.0時出了什麼問題嗎？
下面是一個重現問題的例子。具有以下ARGS運行：
--project=<project-id>
--runner=DirectRunner
--tempLocation=gs://<your-bucket>
--stagingLocation=gs://<your-bucket>
2.0.0-β3（運行正常）
public class EmptySideOutputNPE implements Serializable {
private static final TupleTag<TableRow> mainOutputTag = new TupleTag<TableRow>("mainOutputTag") {
};
private static final TupleTag<TableRow> sideOutputTag = new TupleTag<TableRow>("sideOutputTag") {
};
private static final TupleTag<TableRow> possibleEmptySideOutputTag = new TupleTag<TableRow>("possibleEmptySideOutputTag") {
};
public static void main(String[] args) {
PipelineOptions options = PipelineOptionsFactory
.fromArgs(args)
.withValidation()
.as(PipelineOptions.class);
Pipeline pipeline = Pipeline.create(options);
//Read from BigQuery public dataset
PCollectionTuple results = pipeline.apply("Read-BQ", BigQueryIO.Read.from("bigquery-samples:wikipedia_benchmark.Wiki1k"))
.apply(ParDo.of(new DoFn<TableRow, TableRow>() {
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
TableRow inputRow = c.element();
//output the title to main output tag
TableRow titleRow = new TableRow();
titleRow.set("col", inputRow.get("title"));
c.output(titleRow);
//output the language to the side output
TableRow languageRow = new TableRow();
languageRow.set("col", inputRow.get("language"));
c.sideOutput(sideOutputTag, languageRow);
//don' output anything for the possibleEmptySideOutputTag tag
}
}).withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag).and(possibleEmptySideOutputTag)));
//write the results:
results.get(mainOutputTag).apply("Title write",
BigQueryIO.Write.to("<project-id>:<dataset>.2_0_0_sdk_test_title")
.withCreateDisposition(CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withSchema(getTableSchema()));
results.get(sideOutputTag).apply("Language write",
BigQueryIO.Write.to("<project-id>:<dataset>.2_0_0_sdk_test_language")
.withCreateDisposition(CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withSchema(getTableSchema()));
results.get(possibleEmptySideOutputTag).apply("Empty write",
BigQueryIO.Write.to("<project-id>:<dataset>.2_0_0_sdk_test_empty")
.withCreateDisposition(CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withSchema(getTableSchema()));
pipeline.run();
}
private static TableSchema getTableSchema() {
List<TableFieldSchema> fields = new ArrayList<>();
fields.add(new TableFieldSchema().setName("col").setType("STRING"));
return new TableSchema().setFields(fields);
}
}
2.0.0（NPE）
public class EmptySideOutputNPE implements Serializable {
private static final TupleTag<TableRow> mainOutputTag = new TupleTag<TableRow>("mainOutputTag") {
};
private static final TupleTag<TableRow> sideOutputTag = new TupleTag<TableRow>("sideOutputTag") {
};
private static final TupleTag<TableRow> possibleEmptySideOutputTag = new TupleTag<TableRow>("possibleEmptySideOutputTag") {
};
public static void main(String[] args) {
PipelineOptions options = PipelineOptionsFactory
.fromArgs(args)
.withValidation()
.as(PipelineOptions.class);
Pipeline pipeline = Pipeline.create(options);
//Read from BigQuery public dataset
PCollectionTuple results = pipeline.apply("Read-BQ", BigQueryIO.read().from("bigquery-samples:wikipedia_benchmark.Wiki1k"))
.apply(ParDo.of(new DoFn<TableRow, TableRow>() {
@ProcessElement
public void processElement(ProcessContext c) throws Exception {
TableRow inputRow = c.element();
//output the title to main output tag
TableRow titleRow = new TableRow();
titleRow.set("col", inputRow.get("title"));
c.output(titleRow);
//output the language to the side output
TableRow languageRow = new TableRow();
languageRow.set("col", inputRow.get("language"));
c.output(sideOutputTag, languageRow);
//don' output anything for the possibleEmptySideOutputTag tag
}
}).withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag).and(possibleEmptySideOutputTag)));
//write the results:
results.get(mainOutputTag).apply("Title write",
BigQueryIO.writeTableRows().to("<project-id>:<dataset>.2_0_0_sdk_test_title")
.withCreateDisposition(CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withSchema(getTableSchema()));
results.get(sideOutputTag).apply("Language write",
BigQueryIO.writeTableRows().to("<project-id>:<dataset>.2_0_0_sdk_test_language")
.withCreateDisposition(CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withSchema(getTableSchema()));
results.get(possibleEmptySideOutputTag).apply("Empty write",
BigQueryIO.writeTableRows().to("<project-id>:<dataset>.2_0_0_sdk_test_empty")
.withCreateDisposition(CREATE_IF_NEEDED)
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
.withSchema(getTableSchema()));
pipeline.run();
}
private static TableSchema getTableSchema() {
List<TableFieldSchema> fields = new ArrayList<>();
fields.add(new TableFieldSchema().setName("col").setType("STRING"));
return new TableSchema().setFields(fields);
}
}
23:43:09,484 0 [main] INFO org.apache.beam.sdk.io.gcp.bigquery.BigQuerySourceBase - Starting BigQuery extract job: beam_job_885a1329f1a045d6a6422c975690967e_emptysideoutputnpepolleyg0715134309b6259542-extract
23:43:11,209 1725 [main] INFO org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl - Started BigQuery job: {jobId=beam_job_885a1329f1a045d6a6422c975690967e_emptysideoutputnpepolleyg0715134309b6259542-extract, projectId=<redacted>}.
bq show -j --format=prettyjson --project_id=<redacted> beam_job_885a1329f1a045d6a6422c975690967e_emptysideoutputnpepolleyg0715134309b6259542-extract
23:43:12,718 3234 [main] INFO org.apache.beam.sdk.io.gcp.bigquery.BigQuerySourceBase - BigQuery extract job completed: beam_job_885a1329f1a045d6a6422c975690967e_emptysideoutputnpepolleyg0715134309b6259542-extract
23:43:14,738 5254 [direct-runner-worker] INFO org.apache.beam.sdk.io.FileBasedSource - Matched 1 files for pattern gs://nonsense/BigQueryExtractTemp/885a1329f1a045d6a6422c975690967e/000000000000.avro
23:43:18,171 8687 [direct-runner-worker] INFO org.apache.beam.sdk.io.FileBasedSource - Filepattern gs://nonsense/BigQueryExtractTemp/885a1329f1a045d6a6422c975690967e/000000000000.avro matched 1 files with total size 60370
23:43:18,653 9169 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/956c7d7b866941aaa406bd9e5cb63aab/399d59ec-2475-4d07-9fa9-25feadf53737.
23:43:18,653 9169 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/4377160da6184249a5ffc7cc27155265/8db1d8c4-9e4d-4093-8b9f-3e892de78057.
23:43:22,839 13355 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/956c7d7b866941aaa406bd9e5cb63aab/1b544d4b-650c-4e05-abc0-f80318278a2f.
23:43:22,849 13365 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/4377160da6184249a5ffc7cc27155265/2f3164e0-674e-4926-925f-678657587e75.
23:43:27,428 17944 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/4377160da6184249a5ffc7cc27155265/b0d8ae7a-e6b0-48ac-a0a1-fd3e0fa17f75.
23:43:27,434 17950 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/956c7d7b866941aaa406bd9e5cb63aab/b77b17e3-562c-47b0-8a6c-ee8eb7745fc8.
23:43:33,242 23758 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter - Opening TableRowWriter to gs://nonsense/BigQueryWriteTemp/1f559dd752eb43f7bd1af1c881c21235/a8e51a20-408d-4628-abf3-bbdb2ebd9527.
23:43:35,046 25562 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl - Started BigQuery job: {jobId=956c7d7b866941aaa406bd9e5cb63aab_e9f0a5890698d99399a6106c26d65de2_00001-0, projectId=<redacted>}.
bq show -j --format=prettyjson --project_id=<redacted> 956c7d7b866941aaa406bd9e5cb63aab_e9f0a5890698d99399a6106c26d65de2_00001-0
23:43:35,126 25642 [direct-runner-worker] INFO org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl - Started BigQuery job: {jobId=4377160da6184249a5ffc7cc27155265_a6c30233d929e6958a536246c31fe3d1_00001-0, projectId=<redacted>}.
bq show -j --format=prettyjson --project_id=<redacted> 4377160da6184249a5ffc7cc27155265_a6c30233d929e6958a536246c31fe3d1_00001-0
Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.NullPointerException
at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:322)
at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:292)
at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:200)
at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:63)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:295)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:281)
at com.pipelines.EmptySideOutputNPE.main(EmptySideOutputNPE.java:85)
Caused by: java.lang.NullPointerException
at org.apache.beam.sdk.io.gcp.bigquery.WriteTables.processElement(WriteTables.java:97)
觀察：
- 它從管線上拆下
possibleEmptySideOutputTag當運行在2.0.0精細即.withOutputTags(mainOutputTag, TupleTagList.of(sideOutputTag))); - 加入1+行，當它運行罰款2.0.0到
ParDo中的possibleEmptySideOutputTag。
我應該能夠找到這個，考慮到我評論了原始問題！感謝百萬@jkff。是否有2.0.1的ETA？ –
2.1.0的發佈過程已經開始，所以我會說在一兩個星期內最有可能？ – jkff
輝煌。謝謝。 –