java - 从 PCollection 获取单个字段

Question

我正在使用 Apache Beam 读取数据流并写入 BigQuery，在 tableA 中.我的行存储在 <TableRow> 类型的数据集中.

目前我正在按原样读取行并写入表。但是我想根据 timestamp 过滤行列并拆分结果中的数据 Name FirstName 中的列和 LastName , 在将具有新架构的数据写入名为 tableB 的新表之前

我不知道如何从 PCollection 中获取单个字段数据集，我正在寻求有关获取行字段的正确语法的帮助 Name来 self 的 PCollection<TableRow>

这是我的代码:

PCollection<TableRow> rows =
                        transformedRows.apply("Get rows", BeamIO.getRows());
    
       /*Split a row here and name it rowsAfterColumnSplit
         --
         --
       */
        
        //Write the original set of rows
        WriteResult writeResult1 =
                        rows.apply("write rows",
                                BeamIO.getBigQueryIOWriter(schema, "tableA"));
    
        //Write rowsAfterColumnSplit with new schema
        WriteResult writeResult2 =
                        rowsAfterColumnSplit.apply("write rows after column split",
                                BeamIO.getBigQueryIOWriter(newSchema, "tableB"));

这是示例数据:

| timestamp                    | Name           | City    |

|  2020-07-14 20:12:01.342 UTC | Karl Streisand | Berlin  | 
|  2020-07-14 22:10:10.234 UTC | Anna Karlstad  | Munich  | 

我想通过过滤时间戳在 22.00.00 之后的行来拆分名称“Anna Karlstad”

Answer 1

这里是你如何做到的:

PCollection<TableRow> transformedRows = rows.apply(ParDo.of(Transform.splitColumn()));

这里是 Transform 类:

import com.google.api.services.bigquery.model.TableRow;
import org.apache.beam.sdk.transforms.DoFn;

public class Transform extends DoFn<TableRow, TableRow> {

    private Transform() {
    }

    public static Transform splitColumn() {
        return new Transform();
    }

    @ProcessElement
    public void processElement(@Element TableRow input, OutputReceiver<TableRow> output) {


        if(input.get("Name").toString()!=null){
            input.set("FirstName", input.get("Name").toString().split(" ")[0]);
            input.set("LastName", input.get("Name").toString().split(" ")[1]);
        }
        output.output(input);

    }
}

然后您创建一个包含额外字段的新模式，即 FirstName 和 LastName 并使用 WriteResult 写入 BigQuery