Get individual field from PCollection<TableRow>

Admin 1
Viewed 962

I am reading a stream of data using Apache Beam and writing to BigQuery, in tableA. My rows are stored in dataset of type <TableRow>.

Currently I am reading rows as is and writing to the table. However I would like to filter the rows based on the timestamp column and split data in resulting Name column in FirstName and LastName, before writing the data with the new schema to a new table named tableB

I don´t know how to get individual fields from a PCollection dataset and I am looking for help with the right syntax for getting a row field Name from my PCollection<TableRow>

Here´s my code:

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"));

Here´s sample data:

| 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  |

I want to split name "Anna Karlstad" by filtering rows with timestamp after 22.00.00

apache-beam java
1 Answers

Here´s how you can do that:

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

And here´s the Transform class:

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);

    }
}

You then create a new schema with additional fields i.e FirstName and LastName and write to BigQuery using WriteResult

user_18182860
Related