在 PCollection 分区中传递边输入

Question

我想在 PCollection 分区中传递一个 sideInput，在此基础上，我需要划分我的 PCollection 是他们的无论如何......

PCollectionList<TableRow> part = merged.apply(Partition.of(Pcollection Count Function Called, new PartitionFn<TableRow>(){

                                        @Override
                                        public int partitionFor(TableRow arg0, int arg1) {

                                            return 0;
                                        }

                                    }));

我可以通过任何其他方式对我的 PCollection 进行分区

//不使用动态目标分区 BigQuery 表

merge.apply("write into target", BigQueryIO.writeTableRows()
                                         .to(new SerializableFunction<ValueInSingleWindow<TableRow>, TableDestination>() {
                                                @Override
                                                public TableDestination apply(ValueInSingleWindow<TableRow> value) {
                                                       TableRow row = value.getValue();
                                                       TableReference reference = new TableReference();
                                                       reference.setProjectId("XYZ");
                                                       reference.setDatasetId("ABC");
                                                       System.out.println("date of row " + row.get("authorized_transaction_date_yyyymmdd").toString());               
                                                       LOG.info("date of row "+
                                                       row.get("authorized_transaction_date_yyyymmdd").toString());
                                                       String str = row.get("authorized_transaction_date_yyyymmdd").toString();
                                                       str = str.substring(0, str.length() - 2) + "01";
                                                       System.out.println("str value " + str);
                                                       LOG.info("str value " + str);
                                                       reference.setTableId("TargetTable$" + str);
                                                       return new TableDestination(reference, null);
                                                }
                                         }).withFormatFunction(new SerializableFunction<TableRow, TableRow>() {
                                                @Override
                                                public TableRow apply(TableRow input) {
                                                       LOG.info("format function:"+input.toString());

                                                       return input;
                                                }
                                         })

                                         .withSchema(schema1).withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
                                         .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

现在我必须使用 Dynamic Destination Any Solution.Insted 这个并且必须做分区。

Answer 1

根据您在代码中看到的TableRow，我怀疑您想将PCollection 写入BigQuery，将不同​​的元素发送到不同的BigQuery 表。 BigQueryIO.write() 已经提供了一种方法来做到这一点，使用 BigQueryIO.write().to(DynamicDestinations)。见Writing different values to different BigQuery tables in Apache Beam。