Saya menerima pesan melalui pub-sub dan ingin mengunggah ke kueri besar menggunakan data pesan untuk menentukan tabel tempat data akan diunggah.
Saya mencoba melakukan hal berikut:
Pipa saluran = Pipeline.create(opsi); String bigQueryTable;
PCollection<String> input = pipeline
.apply(PubsubIO.Read.subscription("projects/my-data-analysis/subscriptions/myDataflowSub"));
input.apply(ParDo.of(new DoFn<String, TableRow>() {
@Override
public void processElement(DoFn<String, TableRow>.ProcessContext c) throws Exception {
JSONObject firstJSONObject = new JSONObject(c.element());
bigQueryTable = firstJSONObject.get("tableName").toString();
TableRow tableRow = convertJsonToTableRow(firstJSONObject);
c.output(tableRow);
}
})).apply(BigQueryIO.Write.to("my-data-analysis:mydataset." + bigQueryTable).withSchema(tableSchema));
apakah ada cara melakukan ini tanpa menulis DOFN saya sendiri?
Jika saya perlu menerapkan doFn saya sendiri, bagaimana cara mengimplementasikannya untuk mengunggah ke kueri besar?
