beam_sql_part_2

beam_sql_part_2.java

[..]
public static final Schema SCHEMA = Schema.builder()
            .addStringField("lang")
            .addInt32Field("views")
            .build();
[..]
.apply("transform_to_row", ParDo.of(new RowParDo())).setRowSchema(SCHEMA)
[..]
//ParDo for String -> Row (SQL)
public static class RowParDo extends DoFn<String, Row> {
  @ProcessElement
  public void processElement(ProcessContext c) {
    if (!c.element().equalsIgnoreCase(HEADER)) {
      String[] vals = c.element().split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)");
      Row appRow = Row.withSchema(SCHEMA)
                      .addValues(vals[4], Integer.valueOf(vals[6]))
                      .build();
      c.output(appRow);
     }
   }
}