@Test public void shouldPopulateMissingFeatureValuesWithDefaultInstance() { FeatureRow featureRowWithDuplicatedFeatureFields = createFeatureRow( "myproject/feature_set", Timestamp.newBuilder().setSeconds(10).build(), field("entity_id_primary", 1, Enum.INT32), field("entity_id_secondary", "a", Enum.STRING), field("feature_1", "strValue1", Enum.STRING)); RedisKey expectedKey = createRedisKey( "myproject/feature_set", field("entity_id_primary", 1, Enum.INT32), field("entity_id_secondary", "a", Enum.STRING)); FeatureRow expectedValue = createFeatureRow( "", Timestamp.newBuilder().setSeconds(10).build(), field(hash("feature_1"), "strValue1", Enum.STRING), Field.newBuilder() .setName(hash("feature_2")) .setValue(Value.getDefaultInstance()) .build()); p.apply(Create.of(featureRowWithDuplicatedFeatureFields)).apply(redisFeatureSink.writer()); p.run(); byte[] actual = sync.get(expectedKey.toByteArray()); assertThat(actual, equalTo(expectedValue.toByteArray())); }
@Override public PTransform<PCollection<FeatureRow>, WriteResult> writer() { int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS; if (getRedisClusterConfig() != null) { if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyPrefixSerializer(getRedisClusterConfig().getKeyPrefix()); return new RedisCustomIO.Write( new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else if (getRedisConfig() != null) { if (getRedisConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyProtoSerializer(); return new RedisCustomIO.Write( new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else { throw new RuntimeException( "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink"); } }
RedisFeatureSink implements FeatureSink { @Override public PTransform<PCollection<FeatureRow>, WriteResult> writer() { int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS; if (getRedisClusterConfig() != null) { if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyPrefixSerializer(getRedisClusterConfig().getKeyPrefix()); return new RedisCustomIO.Write( new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else if (getRedisConfig() != null) { if (getRedisConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyProtoSerializer(); return new RedisCustomIO.Write( new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else { throw new RuntimeException( "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink"); } } }
RedisFeatureSink implements FeatureSink { @Override public PTransform<PCollection<FeatureRow>, WriteResult> writer() { int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS; if (getRedisClusterConfig() != null) { if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyPrefixSerializer(getRedisClusterConfig().getKeyPrefix()); return new RedisCustomIO.Write( new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else if (getRedisConfig() != null) { if (getRedisConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyProtoSerializer(); return new RedisCustomIO.Write( new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else { throw new RuntimeException( "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink"); } }  }
RedisFeatureSink implements FeatureSink { @Override public PTransform<PCollection<FeatureRow>, WriteResult> writer() { int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS; if (getRedisClusterConfig() != null) { if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyPrefixSerializer(getRedisClusterConfig().getKeyPrefix()); return new RedisCustomIO.Write( new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else if (getRedisConfig() != null) { if (getRedisConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyProtoSerializer(); return new RedisCustomIO.Write( new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else { throw new RuntimeException( "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink"); } }  static FeatureSink fromConfig(RedisConfig redisConfig); static FeatureSink fromConfig(RedisClusterConfig redisConfig); @Nullable abstract RedisConfig getRedisConfig(); @Nullable abstract RedisClusterConfig getRedisClusterConfig(); abstract Builder toBuilder(); static Builder builder(); RedisFeatureSink withSpecsView(
      PCollectionView<Map<String, Iterable<FeatureSetSpec>>> specsView); @Override PCollection<FeatureSetReference> prepareWrite(
      PCollection<KV<FeatureSetReference, FeatureSetProto.FeatureSetSpec>> featureSetSpecs); @Override PTransform<PCollection<FeatureRow>, WriteResult> writer(); }
RedisFeatureSink implements FeatureSink { @Override public PTransform<PCollection<FeatureRow>, WriteResult> writer() { int flushFrequencySeconds = DEFAULT_FREQUENCY_SECONDS; if (getRedisClusterConfig() != null) { if (getRedisClusterConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisClusterConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyPrefixSerializer(getRedisClusterConfig().getKeyPrefix()); return new RedisCustomIO.Write( new RedisClusterIngestionClient(getRedisClusterConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else if (getRedisConfig() != null) { if (getRedisConfig().getFlushFrequencySeconds() > 0) { flushFrequencySeconds = getRedisConfig().getFlushFrequencySeconds(); } RedisKeySerializer serializer = new RedisKeyProtoSerializer(); return new RedisCustomIO.Write( new RedisStandaloneIngestionClient(getRedisConfig()), getSpecsView(), serializer) .withFlushFrequency(Duration.standardSeconds(flushFrequencySeconds)) .withBatchSize(DEFAULT_BATCH_SIZE); } else { throw new RuntimeException( "At least one RedisConfig or RedisClusterConfig must be provided to Redis Sink"); } }  static FeatureSink fromConfig(RedisConfig redisConfig); static FeatureSink fromConfig(RedisClusterConfig redisConfig); @Nullable abstract RedisConfig getRedisConfig(); @Nullable abstract RedisClusterConfig getRedisClusterConfig(); abstract Builder toBuilder(); static Builder builder(); RedisFeatureSink withSpecsView(
      PCollectionView<Map<String, Iterable<FeatureSetSpec>>> specsView); @Override PCollection<FeatureSetReference> prepareWrite(
      PCollection<KV<FeatureSetReference, FeatureSetProto.FeatureSetSpec>> featureSetSpecs); @Override PTransform<PCollection<FeatureRow>, WriteResult> writer();  }