From 45f67774935f5fe93f85599e4bd79175c8fa0011 Mon Sep 17 00:00:00 2001 From: Tim Sell Date: Mon, 31 Dec 2018 16:03:49 +1100 Subject: [PATCH 1/3] fix imports and do not get gcp application default creds, leave it up to the environment --- ingestion/src/main/java/feast/ingestion/ImportJob.java | 10 ++-------- .../java/feast/ingestion/options/ImportJobOptions.java | 4 +--- 2 files changed, 3 insertions(+), 11 deletions(-) diff --git a/ingestion/src/main/java/feast/ingestion/ImportJob.java b/ingestion/src/main/java/feast/ingestion/ImportJob.java index e4c1f7fb812..579eaef9cf5 100644 --- a/ingestion/src/main/java/feast/ingestion/ImportJob.java +++ b/ingestion/src/main/java/feast/ingestion/ImportJob.java @@ -18,8 +18,6 @@ package feast.ingestion; import com.google.api.services.bigquery.model.TableRow; -import com.google.api.services.dataflow.DataflowScopes; -import com.google.auth.oauth2.GoogleCredentials; import com.google.inject.Guice; import com.google.inject.Inject; import com.google.inject.Injector; @@ -109,15 +107,11 @@ public static void main(String[] args) { public static PipelineResult mainWithResult(String[] args) { log.info("Arguments: " + Arrays.toString(args)); - ImportJobOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(ImportJobOptions.class); + ImportJobOptions options = + PipelineOptionsFactory.fromArgs(args).withValidation().as(ImportJobOptions.class); if (options.getJobName().isEmpty()) { options.setJobName(generateName()); } - try { - options.setGcpCredential(GoogleCredentials.getApplicationDefault().createScoped(DataflowScopes.all())); - } catch (IOException e) { - log.error("Exception while setting gcp credential manually : ", e.getMessage()); - } log.info("options: " + options.toString()); ImportSpec importSpec = new ImportSpecSupplier(options).get(); Injector injector = diff --git a/ingestion/src/main/java/feast/ingestion/options/ImportJobOptions.java b/ingestion/src/main/java/feast/ingestion/options/ImportJobOptions.java index 687a8213e57..8ec34df5251 100644 --- a/ingestion/src/main/java/feast/ingestion/options/ImportJobOptions.java +++ b/ingestion/src/main/java/feast/ingestion/options/ImportJobOptions.java @@ -19,8 +19,6 @@ import com.google.auto.service.AutoService; import java.util.Collections; -import org.apache.beam.runners.flink.FlinkPipelineOptions; -import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; import org.apache.beam.sdk.metrics.MetricsSink; import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description; @@ -28,7 +26,7 @@ import org.apache.beam.sdk.options.PipelineOptionsRegistrar; import org.apache.beam.sdk.options.Validation.Required; -public interface ImportJobOptions extends PipelineOptions, FlinkPipelineOptions, GcpOptions { +public interface ImportJobOptions extends PipelineOptions { @Description("Import spec yaml file path") @Required(groups = {"importSpec"}) String getImportSpecYamlFile(); From f019e618d28741cbc84131deeb8b91e29585ada3 Mon Sep 17 00:00:00 2001 From: Tim Sell Date: Mon, 31 Dec 2018 16:05:22 +1100 Subject: [PATCH 2/3] use KafkaReadOptions class, to parse import spec options map --- .../transform/FeatureRowKafkaIO.java | 115 +++++++++--------- 1 file changed, 56 insertions(+), 59 deletions(-) diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java index e8b8becfd96..eeb282759dc 100644 --- a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java +++ b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java @@ -17,14 +17,19 @@ package feast.ingestion.transform; -import com.google.common.base.Preconditions; -import com.google.common.base.Strings; +import static com.google.common.base.Preconditions.checkArgument; + import feast.ingestion.deserializer.FeatureRowDeserializer; import feast.ingestion.deserializer.FeatureRowKeyDeserializer; +import feast.options.Options; import feast.options.OptionsParser; import feast.specs.ImportSpecProto.ImportSpec; import feast.types.FeatureRowProto.FeatureRow; import feast.types.FeatureRowProto.FeatureRowKey; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import javax.validation.constraints.NotEmpty; import org.apache.beam.sdk.io.kafka.KafkaIO; import org.apache.beam.sdk.io.kafka.KafkaRecord; import org.apache.beam.sdk.transforms.DoFn; @@ -32,69 +37,61 @@ import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PInput; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -import static com.google.common.base.Preconditions.checkArgument; - public class FeatureRowKafkaIO { + static final String KAFKA_TYPE = "kafka"; - static final String KAFKA_TYPE = "kafka"; + /** + * Transform for reading {@link feast.types.FeatureRowProto.FeatureRow FeatureRow} proto messages + * from kafka one or more kafka topics. + */ + public static Read read(ImportSpec importSpec) { + return new Read(importSpec); + } + public static class KafkaReadOptions implements Options { + @NotEmpty public String server; + @NotEmpty public String topics; + } - /** - * Transform for reading {@link feast.types.FeatureRowProto.FeatureRow FeatureRow} - * proto messages from kafka one or more kafka topics. - * - */ - public static Read read(ImportSpec importSpec) { - return new Read(importSpec); - } + public static class Read extends FeatureIO.Read { - public static class Read extends FeatureIO.Read { + private ImportSpec importSpec; - private ImportSpec importSpec; - - private Read(ImportSpec importSpec) { - this.importSpec = importSpec; - } - - @Override - public PCollection expand(PInput input) { - - checkArgument(importSpec.getType().equals(KAFKA_TYPE)); - - String bootstrapServer = importSpec.getOptionsMap().get("server"); - - Preconditions.checkArgument( - !Strings.isNullOrEmpty(bootstrapServer), "kafka bootstrap server must be set"); - - String topics = importSpec.getOptionsMap().get("topics"); - - Preconditions.checkArgument( - !Strings.isNullOrEmpty(topics), "kafka topic(s) must be set"); - - List topicsList = new ArrayList<>(Arrays.asList(topics.split(","))); - - KafkaIO.Read kafkaIOReader = KafkaIO.read() - .withBootstrapServers(bootstrapServer) - .withTopics(topicsList) - .withKeyDeserializer(FeatureRowKeyDeserializer.class) - .withValueDeserializer(FeatureRowDeserializer.class); - - PCollection> featureRowRecord = input.getPipeline().apply(kafkaIOReader); + private Read(ImportSpec importSpec) { + this.importSpec = importSpec; + } - PCollection featureRow = featureRowRecord.apply( - ParDo.of( - new DoFn, FeatureRow>() { - @ProcessElement - public void processElement(ProcessContext processContext) { - KafkaRecord record = processContext.element(); - processContext.output(record.getKV().getValue()); - } - })); - return featureRow; - } + @Override + public PCollection expand(PInput input) { + + checkArgument(importSpec.getType().equals(KAFKA_TYPE)); + + KafkaReadOptions options = + OptionsParser.parse(importSpec.getOptionsMap(), KafkaReadOptions.class); + + List topicsList = new ArrayList<>(Arrays.asList(options.topics.split(","))); + + KafkaIO.Read kafkaIOReader = + KafkaIO.read() + .withBootstrapServers(options.server) + .withTopics(topicsList) + .withKeyDeserializer(FeatureRowKeyDeserializer.class) + .withValueDeserializer(FeatureRowDeserializer.class); + + PCollection> featureRowRecord = + input.getPipeline().apply(kafkaIOReader); + + PCollection featureRow = + featureRowRecord.apply( + ParDo.of( + new DoFn, FeatureRow>() { + @ProcessElement + public void processElement(ProcessContext processContext) { + KafkaRecord record = processContext.element(); + processContext.output(record.getKV().getValue()); + } + })); + return featureRow; } + } } From 47b6d032ca3f213dd445003e2bbbf23caded39a6 Mon Sep 17 00:00:00 2001 From: Tim Sell Date: Mon, 31 Dec 2018 16:06:25 +1100 Subject: [PATCH 3/3] reformat using the google-java-format plugin --- .../deserializer/FeatureRowDeserializer.java | 29 ++-- .../FeatureRowKeyDeserializer.java | 32 ++-- .../KafkaFeatureRowDeserializerTest.java | 163 +++++++++--------- 3 files changed, 105 insertions(+), 119 deletions(-) diff --git a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java b/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java index f83c6fb0f6e..5abff1b2bb3 100644 --- a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java +++ b/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java @@ -2,11 +2,10 @@ import com.google.protobuf.InvalidProtocolBufferException; import feast.types.FeatureRowProto.FeatureRow; +import java.util.Map; import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.serialization.Deserializer; -import java.util.Map; - /** * Deserializer for Kafka to deserialize Protocol Buffers messages * @@ -14,20 +13,18 @@ */ public class FeatureRowDeserializer implements Deserializer { - @Override - public void configure(Map configs, boolean isKey) { - } + @Override + public void configure(Map configs, boolean isKey) {} - @Override - public FeatureRow deserialize(String topic, byte[] data) { - try { - return FeatureRow.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new SerializationException("Error deserializing FeatureRow from Protobuf message", e); - } + @Override + public FeatureRow deserialize(String topic, byte[] data) { + try { + return FeatureRow.parseFrom(data); + } catch (InvalidProtocolBufferException e) { + throw new SerializationException("Error deserializing FeatureRow from Protobuf message", e); } + } - @Override - public void close() { - } -} \ No newline at end of file + @Override + public void close() {} +} diff --git a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java b/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java index 74c5b03fd9d..01ca9c70686 100644 --- a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java +++ b/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java @@ -1,12 +1,11 @@ package feast.ingestion.deserializer; import com.google.protobuf.InvalidProtocolBufferException; -import feast.types.FeatureRowProto.*; +import feast.types.FeatureRowProto.FeatureRowKey; +import java.util.Map; import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.serialization.Deserializer; -import java.util.Map; - /** * Deserializer for Kafka to deserialize Protocol Buffers messages * @@ -14,20 +13,19 @@ */ public class FeatureRowKeyDeserializer implements Deserializer { - @Override - public void configure(Map configs, boolean isKey) { - } + @Override + public void configure(Map configs, boolean isKey) {} - @Override - public FeatureRowKey deserialize(String topic, byte[] data) { - try { - return FeatureRowKey.parseFrom(data); - } catch (InvalidProtocolBufferException e) { - throw new SerializationException("Error deserializing FeatureRowKey from Protobuf message", e); - } + @Override + public FeatureRowKey deserialize(String topic, byte[] data) { + try { + return FeatureRowKey.parseFrom(data); + } catch (InvalidProtocolBufferException e) { + throw new SerializationException( + "Error deserializing FeatureRowKey from Protobuf message", e); } + } - @Override - public void close() { - } -} \ No newline at end of file + @Override + public void close() {} +} diff --git a/ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java b/ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java index c6fc30d1366..1435778446d 100644 --- a/ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java +++ b/ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java @@ -1,8 +1,12 @@ package feast.ingestion.deserializer; - import com.google.protobuf.MessageLite; import feast.types.FeatureRowProto.FeatureRow; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.LinkedBlockingQueue; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.KafkaException; @@ -14,7 +18,11 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.core.*; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.listener.MessageListener; @@ -27,101 +35,84 @@ import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.concurrent.ListenableFuture; -import java.util.Map; -import java.util.UUID; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.LinkedBlockingQueue; - @RunWith(SpringRunner.class) @EmbeddedKafka(controlledShutdown = true) public class KafkaFeatureRowDeserializerTest { - @Configuration - static class ContextConfiguration { + @Autowired private EmbeddedKafkaBroker embeddedKafka; + @Autowired private KafkaTemplate template; - @Autowired - private EmbeddedKafkaBroker embeddedKafka; + private void deserialize(MessageType input) { + // generate a random UUID to create a unique topic and consumer group id for each test + String uuid = UUID.randomUUID().toString(); + String topic = "topic-" + uuid; - @Bean - ProducerFactory producerFactory() { - Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + embeddedKafka.addTopics(topic); - return new DefaultKafkaProducerFactory<>(producerProps, - new ByteArraySerializer(), - new ByteArraySerializer()); - } + Deserializer deserializer = new FeatureRowDeserializer(); - @Bean - KafkaTemplate kafkaTemplate() { - return new KafkaTemplate<>( - producerFactory(), - true); - } + Map consumerProps = + KafkaTestUtils.consumerProps(uuid, Boolean.FALSE.toString(), embeddedKafka); + ConsumerFactory consumerFactory = + new DefaultKafkaConsumerFactory<>(consumerProps, deserializer, deserializer); + + BlockingQueue> records = new LinkedBlockingQueue<>(); + ContainerProperties containerProps = new ContainerProperties(topic); + containerProps.setMessageListener((MessageListener) records::add); + + MessageListenerContainer container = + new KafkaMessageListenerContainer<>(consumerFactory, containerProps); + container.start(); + ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); + + byte[] data = input.toByteArray(); + ProducerRecord producerRecord = new ProducerRecord<>(topic, data, data); + ListenableFuture> producerFuture = template.send(producerRecord); + + try { + producerFuture.get(); + } catch (InterruptedException e) { + return; + } catch (ExecutionException e) { + throw new KafkaException("Error sending message to Kafka.", e.getCause()); + } + + ConsumerRecord consumerRecord; + try { + consumerRecord = records.take(); + } catch (InterruptedException e) { + return; } - @Autowired - private EmbeddedKafkaBroker embeddedKafka; - - @Autowired - private KafkaTemplate template; - - private void deserialize(MessageType input) { - // generate a random UUID to create a unique topic and consumer group id for each test - String uuid = UUID.randomUUID().toString(); - String topic = "topic-" + uuid; - - embeddedKafka.addTopics(topic); - - Deserializer deserializer = new FeatureRowDeserializer(); - - Map consumerProps = KafkaTestUtils.consumerProps( - uuid, Boolean.FALSE.toString(), embeddedKafka); - ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>( - consumerProps, - deserializer, deserializer); - - BlockingQueue> records = new LinkedBlockingQueue<>(); - ContainerProperties containerProps = new ContainerProperties(topic); - containerProps.setMessageListener((MessageListener) records::add); - - MessageListenerContainer container = new KafkaMessageListenerContainer<>( - consumerFactory, - containerProps); - container.start(); - ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); - - byte[] data = input.toByteArray(); - ProducerRecord producerRecord = new ProducerRecord<>(topic, data, data); - ListenableFuture> producerFuture = template.send(producerRecord); - - try { - producerFuture.get(); - } catch (InterruptedException e) { - return; - } catch (ExecutionException e) { - throw new KafkaException("Error sending message to Kafka.", e.getCause()); - } - - ConsumerRecord consumerRecord; - try { - consumerRecord = records.take(); - } catch (InterruptedException e) { - return; - } - - FeatureRow key = consumerRecord.key(); - Assert.assertEquals(key, input); - - FeatureRow value = consumerRecord.value(); - Assert.assertEquals(value, input); + FeatureRow key = consumerRecord.key(); + Assert.assertEquals(key, input); + + FeatureRow value = consumerRecord.value(); + Assert.assertEquals(value, input); + } + + @Test(timeout = 10000) + public void deserializeFeatureRowProto() { + FeatureRow message = FeatureRow.newBuilder().setEntityName("test").build(); + deserialize(message); + } + + @Configuration + static class ContextConfiguration { + + @Autowired private EmbeddedKafkaBroker embeddedKafka; + + @Bean + ProducerFactory producerFactory() { + Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + + return new DefaultKafkaProducerFactory<>( + producerProps, new ByteArraySerializer(), new ByteArraySerializer()); } - @Test(timeout = 10000) - public void deserializeFeatureRowProto() { - FeatureRow message = FeatureRow.newBuilder() - .setEntityName("test") - .build(); - deserialize(message); + @Bean + KafkaTemplate kafkaTemplate() { + return new KafkaTemplate<>(producerFactory(), true); } + } }