从 PCollection<TableRow> 获取单个字段答案

【问题标题】：Get individual field from PCollection<TableRow>从 PCollection<TableRow> 获取单个字段
【发布时间】：2020-07-14 19:42:57
【问题描述】：

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

目前我正在读取行并写入表。但是，我想根据timestamp 列过滤行并将数据拆分为FirstName 和LastName 中的Name 列，然后将具有新模式的数据写入名为tableB

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

这是我的代码：

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”

【问题讨论】：

使用ParDo 或Map
发布样本数据
可以粘贴问题中的数据
你的意思是拆分 - “安娜卡尔斯塔德”到["Anna", "Karlstad"]？
没错，@bigbounty。之后，我将创建一个新列来保存每个元素并更新架构。因此生成的模式将包含 5 个字段，即 timestamp, Name, FirstName, LastName, City

标签： java apache-beam

【解决方案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

【讨论】：