@Test public void transformationTest() { KeyValueTestStoreStub<LatLng, GeocodeResponse> kvStore = new KeyValueTestStoreStub<>(); kvStore.put(new LatLng(56.26d, 9.51d), toGeocodeResponse(Country.DENMARK)); kvStore.put(new LatLng(36.21d, 138.25d), toGeocodeResponse(Country.JAPAN)); kvStore.put(new LatLng(88.21d, -32.01d), toGeocodeResponse(null)); SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStore = () -> GeocodeKvStore.create(kvStore); final String[] denmark = { "0", Country.DENMARK.getTitle(), Country.DENMARK.getIso2LetterCode(), "EUROPE", "100.0", "110.0", "111.0", "200.0", "Ocean", "220.0", "222.0", "30.0", "0.00001", "56.26", "9.51", "Copenhagen", "GEODETIC_DATUM_ASSUMED_WGS84", "155.5", "44.5", "105.0", "5.0", "false", "DNK", "DNK.2_1", "DNK.2.14_1", null, "Denmark", "Midtjylland", "Silkeborg", null }; final String[] japan = { "1", Country.JAPAN.getTitle(), Country.JAPAN.getIso2LetterCode(), "ASIA", "100.0", "110.0", "111.0", "200.0", "Ocean", "220.0", "222.0", "30.0", "0.00001", "36.21", "138.25", "Tokyo", "GEODETIC_DATUM_ASSUMED_WGS84", "155.5", "44.5", "105.0", "5.0", "true", "JPN", "JPN.26_1", "JPN.26.40_1", null, "Japan", "Nagano", "Nagawa", null }; final String[] arctic = { "2", null, null, null, "-80.0", "-40.0", "0.0", "5.0", "Arctic Ocean", "0.0", "-1.5", "500.0", "0.01", "88.21", "-32.01", null, "GEODETIC_DATUM_ASSUMED_WGS84", "2.5", "2.5", "-60.0", "20.0", null, null, null, null, null, null, null, null, null }; final MetadataRecord mdr = MetadataRecord.newBuilder() .setId("0") .setDatasetPublishingCountry(Country.DENMARK.getIso2LetterCode()) .setDatasetKey(UUID.randomUUID().toString()) .build(); final List<ExtendedRecord> records = createExtendedRecordList(mdr, denmark, japan, arctic); final List<LocationRecord> locations = createLocationList(mdr, denmark, japan, arctic); PCollectionView<MetadataRecord> metadataView = p.apply("Create test metadata", Create.of(mdr)) .apply("Convert into view", View.asSingleton()); PCollection<LocationRecord> recordCollection = p.apply(Create.of(records)) .apply( LocationTransform.builder() .geocodeKvStoreSupplier(geocodeKvStore) .metadataView(metadataView) .create() .interpret()) .apply("Cleaning Date created", ParDo.of(new RemoveDateCreated())); PAssert.that(recordCollection).containsInAnyOrder(locations); p.run(); }
@Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } @Builder(buildMethodName = "create") private  LocationTransform(
      SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStoreSupplier,
      PCollectionView<MetadataRecord> metadataView); }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } @Builder(buildMethodName = "create") private  LocationTransform(
      SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStoreSupplier,
      PCollectionView<MetadataRecord> metadataView); MapElements<LocationRecord, KV<String, LocationRecord>> toKv(); LocationTransform counterFn(SerializableConsumer<String> counterFn); @Override SingleOutput<ExtendedRecord, LocationRecord> interpret(); @Setup void setup(); @Teardown void tearDown(); @Override Optional<LocationRecord> convert(ExtendedRecord source); @Override @ProcessElement void processElement(ProcessContext c); Optional<LocationRecord> processElement(ExtendedRecord source, MetadataRecord mdr); }
LocationTransform extends Transform<ExtendedRecord, LocationRecord> { @Override public SingleOutput<ExtendedRecord, LocationRecord> interpret() { return ParDo.of(this).withSideInputs(metadataView); } @Builder(buildMethodName = "create") private  LocationTransform(
      SerializableSupplier<KeyValueStore<LatLng, GeocodeResponse>> geocodeKvStoreSupplier,
      PCollectionView<MetadataRecord> metadataView); MapElements<LocationRecord, KV<String, LocationRecord>> toKv(); LocationTransform counterFn(SerializableConsumer<String> counterFn); @Override SingleOutput<ExtendedRecord, LocationRecord> interpret(); @Setup void setup(); @Teardown void tearDown(); @Override Optional<LocationRecord> convert(ExtendedRecord source); @Override @ProcessElement void processElement(ProcessContext c); Optional<LocationRecord> processElement(ExtendedRecord source, MetadataRecord mdr);  }