如何将侧边输入/额外输入传递到JdbcIO RowMapper Java

Question

我正在尝试使用jdbcio.read在java beam中读取cloud SQL表。我想使用.WithRowMapper（Resultset Resultset）方法将Resultset中的每一行转换为GenericData.Record。是否有一种方法可以将JSON模式字符串作为输入传递给。WithRowMapper方法，如ParDo接受sideInputs作为PCollectionView

PCollection<String> avroSchema= pipeline.apply(JdbcIO.<String>read()
                .withDataSourceConfiguration(config)
                .withCoder(StringUtf8Coder.of())
                .withQuery("SELECT DISTINCT column_name, data_type \n" +
                        "FROM information_schema.columns\n" +
                        "WHERE table_name = " + "'" + tableName + "'")
                .withRowMapper((JdbcIO.RowMapper<String>) resultSet -> {
            // code here to generate avro schema string
           // this works fine for me

}))

 PCollectionView<String> s = avroSchema.apply(View.<String>asSingleton());

// I want to access this view as side input in next JdbcIO.Read operation
// something like this ;

pipeline.apply(JdbcIO.<String>read()
        .withDataSourceConfiguration(config)
        .withCoder(StringUtf8Coder.of())
        .withQuery(queryString)
        .withRowMapper(new JdbcIO.RowMapper<String>() {

            @Override
            public String mapRow(ResultSet resultSet) throws Exception {
                // access schema here and use it to parse and create 
               //GenericData.Record from ResultSet fields as per schema

                return null;
            }
        })).

    withSideInputs(My PCollectionView here); // this option is not there right now.

Answer 1

此时，IOs API不接受SideInputs。

应该可以在读取和执行映射后立即添加ParDo。ParDo可以接受侧输入。