join example in batch mode using DataStream API

[Etienne Chauchot]

Hi all,

In case it is useful to people:

I was testing new DataStream convergent batch/streaming API and join did
not work in batch mode : https://issues.apache.org/jira/browse/FLINK-22587

I had to manually code an inner join using *KeyedCoProcessFunction* and
*states*. Here is an example of a manual join (implementing part of
TPC-DS query3 with avro GenericRecords) it may not be the best code, but
it could serve as an example for people interested

Best,

Etienne


// Join1: WHERE date_dim.d_date_sk = store_sales.ss_sold_date_sk Schema schemaJoinDateSk =AvroUtils .getSchemaMerged(dateDimAvroSchema, storeSalesAvroSchema, "recordsJoinDateSk"); final DataStream<GenericRecord> recordsJoinDateSk = dateDim
   .keyBy((KeySelector<GenericRecord, Integer>) value -> (Integer) value.get("d_date_sk"))
   .connect(storeSales.keyBy(
     (KeySelector<GenericRecord, Integer>) value -> (Integer) value.get("ss_sold_date_sk")))
   .process(new JoinRecords(dateDimAvroSchema, storeSalesAvroSchema, schemaJoinDateSk))
   .returns(new GenericRecordAvroTypeInfo(schemaJoinDateSk));


Where *JoinRecords* does a manual inner join with states :


   public JoinRecords(Schema schemaLhs, Schema schemaRhs, Schema outputSchema) {
     this.schemaLhs = schemaLhs; this.schemaRhs = schemaRhs; this.outputSchema = outputSchema; this.schemaLhsString = schemaLhs.toString(); this.schemaRhsString = schemaRhs.toString(); this.schemaString = outputSchema.toString(); }

   @Override public void open(Configuration parameters)throws Exception {
     state1 = getRuntimeContext().getMapState(
       new MapStateDescriptor<>("records_dataStream_1", Integer.class, GenericRecord.class)); state2 = getRuntimeContext().getMapState(
       new MapStateDescriptor<>("records_dataStream_2", Integer.class, GenericRecord.class)); }

   private GenericRecordjoinRecords(GenericRecord first, GenericRecord second)
     throws Exception {
     // after deserialization if (schemaLhs ==null) {
       schemaLhs =new Schema.Parser().parse(schemaLhsString); }
     if (schemaRhs ==null) {
       schemaRhs =new Schema.Parser().parse(schemaRhsString); }
     if (outputSchema ==null) {
       outputSchema =new Schema.Parser().parse(schemaString); }

     GenericRecord outputRecord =new GenericRecordBuilder(outputSchema).build(); for (Schema.Field f :outputSchema.getFields()) {
       if (schemaLhs.getField(f.name()) !=null) {
         outputRecord.put(f.name(), first.get(f.name())); }else if (schemaRhs.getField(f.name()) !=null) {
         outputRecord.put(f.name(), second.get(f.name())); }
     }
     return outputRecord; }

   private GenericRecordstateJoin(GenericRecord currentRecord, int currentDatastream, Context context)throws Exception {
     final Integer currentKey = context.getCurrentKey(); MapState<Integer, GenericRecord> myState = currentDatastream ==1 ?state1 :state2; MapState<Integer, GenericRecord> otherState = currentDatastream ==1 ?state2 :state1; // join with the other datastream by looking into the state of
the other datastream final GenericRecord otherRecord = otherState.get(currentKey); if (otherRecord ==null) {// did not find a record to join with, store record for later join myState.put(currentKey, currentRecord); return null; }else {// found a record to join with (same key), join (with using the correct
avro schema) return currentDatastream ==1 ? joinRecords(currentRecord, otherRecord) : joinRecords(otherRecord, currentRecord); }
   }

   @Override public void processElement1(GenericRecord currentRecord, Context context, Collector<GenericRecord> collector)throws Exception {
     final GenericRecord jointRecord = stateJoin(currentRecord, 1, context); if (jointRecord !=null) {
       collector.collect(jointRecord); }
   }

   @Override public void processElement2(GenericRecord currentRecord, Context context, Collector<GenericRecord> collector)throws Exception {
     final GenericRecord jointRecord = stateJoin(currentRecord, 2, context); if (jointRecord !=null) {
       collector.collect(jointRecord); }
   }
}