Dataflow sideinput refresh

Question

We are trying to refresh the sideinput at a certain interval in a streaming dataflow job. Followed this link https://beam.apache.org/documentation/patterns/side-inputs/ But not able to achieve it syntactically also right. Can someone help with the right way of implementing sideinput refresh for the below piece of code?

PCollection<KV<String, String>> updateVariable = pipeline.apply(JdbcIO.<KV<String, String>>read()
            .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration
                    .create("com.mysql.jdbc.Driver", "jdbc:mysql://" + sqlIp + "/" + sqlDb).withUsername(sqlUser)
                    .withPassword(sqlPwd))
            .withQuery(
                    "select * from OBD_LOOKUP")
            .withCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
            .withRowMapper(new JdbcIO.RowMapper<KV<String, String>>() {
                @Override
                public KV<String, String> mapRow(java.sql.ResultSet resultSet) throws Exception {
                    / TODO Auto-generated method stub
                    return KV.of(resultSet.getString(1), resultSet.getString(2));
                }
            }));

   final PCollectionView<Map<String, String>> lookupCollection = updateVariable             
            .apply("Assign into Global Window",
            Window.<KV<String, String>>into(new GlobalWindows())
                    .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
                    .accumulatingFiredPanes())
            .apply("SideInputViewFormed", View.<String, String>asMap());

    PCollection<String> resultnew = result
            .apply(ParDo.of(new ObdLookUpSideInput(lookupCollection)).withSideInputs(lookupCollection));

Which pattern in the guide did you attempt? GenerateSequence or PeriodicImpulse? — 大ドア東

Reza Rokni Reza Rokni · Accepted Answer · 2020-12-02T01:31:04

The pattern makes use of

View.asSinglton()

View.asMap is itself an aggregation, it will inherit triggering and accumulate its input. In this specific pattern, setting the pipeline to accumulatingPanes mode before this transform will result in duplicate key errors.

More detail on this can be found in the answer to this question Solve duplicate values with view as map.

Dataflow sideinput refresh

1 Answers