You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@samza.apache.org by GitBox <gi...@apache.org> on 2020/01/23 18:47:22 UTC
[GitHub] [samza-hello-samza] lakshmi-manasa-g opened a new pull request #71:
Sample for producing to Azure Blob Storage
lakshmi-manasa-g opened a new pull request #71: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71
Feature: Sample Samza job related to SEP-26: Azure Blob System Producer
Changes: New high-level Yarn job "AzureBlobApplication" added to the samples
Tests: Sample job successfully produces blobs for configured Azure Storage Account. Confirmed on Azure portal.
Upgrade instructions: Backwards compatible change as its a completely new job and hence no upgrade needed.
Usage instructions: Add Azure Storage Account details to the configs of the Samza job. Step-by-step instructions to run the job are provided in the javadocs of the job.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] lakshmi-manasa-g commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
lakshmi-manasa-g commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371915951
##########
File path: pom.xml
##########
@@ -206,6 +206,26 @@ under the License.
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
+ <!-- Jackson Dependencies -->
Review comment:
No the app does not need them directly.
But azure-storage-blob depends on azure-core which in-turn needs Jackson.
removing these dependencies throws
`java.lang.NoClassDefFoundError: com/fasterxml/jackson/core/TSFBuilder
at com.fasterxml.jackson.dataformat.xml.XmlMapper.<init>(XmlMapper.java:122) ~[jackson-dataformat-xml-2.10.0.jar:2.10.0]
at com.azure.core.implementation.serializer.jackson.JacksonAdapter.<init>(JacksonAdapter.java:75) ~[azure-core-1.0.0.jar:?]
at com.azure.core.implementation.serializer.jackson.JacksonAdapter.createDefaultSerializerAdapter(JacksonAdapter.java:108) ~[azure-core-1.0.0.jar:?]
at com.azure.core.implementation.RestProxy.createDefaultSerializer(RestProxy.java:629) ~[azure-core-1.0.0.jar:?]
at com.azure.core.implementation.RestProxy.create(RestProxy.java:691) ~[azure-core-1.0.0.jar:?]
at com.azure.storage.blob.implementation.ServicesImpl.<init>(ServicesImpl.java:58) ~[azure-storage-blob-12.0.1.jar:?]
at com.azure.storage.blob.implementation.AzureBlobStorageImpl.<init>(AzureBlobStorageImpl.java:213) ~[azure-storage-blob-12.0.1.jar:?]
at com.azure.storage.blob.implementation.AzureBlobStorageBuilder.build(AzureBlobStorageBuilder.java:90) ~[azure-storage-blob-12.0.1.jar:?]
at com.azure.storage.blob.BlobServiceAsyncClient.<init>(BlobServiceAsyncClient.java:90) ~[azure-storage-blob-12.0.1.jar:?]
at com.azure.storage.blob.BlobServiceClientBuilder.buildAsyncClient(BlobServiceClientBuilder.java:103) ~[azure-storage-blob-12.0.1.jar:?]
at org.apache.samza.system.azureblob.producer.AzureBlobSystemProducer.setupAzureContainer(AzureBlobSystemProducer.java:370) ~[samza-azure_2.11-1.4.903-SNAPSHOT.jar:?]`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371586350
##########
File path: src/main/java/samza/examples/azure/AzureBlobApplication.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package samza.examples.azure;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+import org.apache.samza.operators.MessageStream;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.serializers.JsonSerdeV2;
+import org.apache.samza.serializers.NoOpSerde;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
+import samza.examples.azure.data.PageViewAvroRecord;
+import samza.examples.cookbook.data.PageView;
+
+/**
+ * In this example, we demonstrate sending blobs to Azure Blob Storage.
+ * This Samza job reads from Kafka topic "page-view-azure-blob-input" and produces blobs to Azure-Container "oss-testcontainer" in your Azure Storage account.
+ *
+ * Currently, Samza supports sending Avro files are blobs.
+ * Hence the incoming messages into the Samza job have to be converted to an Avro record.
+ * For this job, we use input message as {@link samza.examples.cookbook.data.PageView} and
+ * covert it to an Avro record defined as {@link samza.examples.azure.data.PageViewAvroRecord}.
+ *
+ * To run the below example:
+ *
+ * <ol>
+ * <li>
+ * Replace your-azure-storage-account-name and your-azure-storage-account-key with details of your Azure Storage Account.
+ * </li>
+ * <li>
+ * Ensure that the topic "page-view-azure-blob-input" is created <br/>
+ * ./deploy/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic page-view-azure-blob-input --partitions 1 --replication-factor 1
+ * </li>
+ * <li>
+ * Run the application using the run-app.sh script <br/>
+ * ./deploy/samza/bin/run-app.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/azure-blob-application.properties
+ * </li>
+ * <li>
+ * Produce some messages to the "page-view-azure-blob-input" topic <br/>
+ * ./deploy/kafka/bin/kafka-console-producer.sh --topic page-view-azure-blob-input --broker-list localhost:9092 <br/>
+ * {"userId": "user1", "country": "india", "pageId":"google.com"} <br/>
+ * {"userId": "user2", "country": "france", "pageId":"facebook.com"} <br/>
+ * {"userId": "user3", "country": "china", "pageId":"yahoo.com"} <br/>
+ * {"userId": "user4", "country": "italy", "pageId":"linkedin.com"} <br/>
+ * {"userId": "user5", "country": "germany", "pageId":"amazon.com"} <br/>
+ * {"userId": "user6", "country": "denmark", "pageId":"apple.com"} <br/>
+ * </li>
+ * <li>
+ * Seeing Output:
+ * <ol>
+ * <li>
+ * See blobs in your Azure portal at https://<azure-storage-account-name>.blob.core.windows.net/oss-testcontainer/PageViewEventStream/<time-stamp>.avro
+ * </li>
+ * <li>
+ * system-name "oss-testcontainer" in configs and code below maps to Azure-Container in Azure Storage account.
+ * </li>
+ * <li>
+ * <time-stamp> is of the format yyyy/MM/dd/HH/mm-ss-randomString.avro. Hence navigate through the virtual folders on the portal to see your blobs.
+ * </li>
+ * <li>
+ * Due to network calls, allow a few minutes for blobs to appear on the portal.
+ * </li>
+ * <li>
+ * Config "maxMessagesPerBlob=2" ensures that a blob is created per 2 input messages. Adjust input or config accordingly.
+ * </li>
+ * </ol>
+ * </li>
+ * </ol>
+ */
+public class AzureBlobApplication implements StreamApplication {
+ private static final List<String> KAFKA_CONSUMER_ZK_CONNECT = ImmutableList.of("localhost:2181");
+ private static final List<String> KAFKA_PRODUCER_BOOTSTRAP_SERVERS = ImmutableList.of("localhost:9092");
+ private static final Map<String, String> KAFKA_DEFAULT_STREAM_CONFIGS = ImmutableMap.of("replication.factor", "1");
+ private static final String INPUT_PAGEVIEW_STREAM_ID = "page-view-azure-blob-input";
+ private static final String OUTPUT_SYSTEM = "oss-testcontainer";
+ private static final String OUTPUT_STREAM = "PageViewEventStream";
+
+ @Override
+ public void describe(StreamApplicationDescriptor appDescriptor) {
+ // Define a system descriptor for Kafka
+ KafkaSystemDescriptor kafkaSystemDescriptor =
+ new KafkaSystemDescriptor("kafka").withConsumerZkConnect(KAFKA_CONSUMER_ZK_CONNECT)
+ .withProducerBootstrapServers(KAFKA_PRODUCER_BOOTSTRAP_SERVERS)
+ .withDefaultStreamConfigs(KAFKA_DEFAULT_STREAM_CONFIGS);
+
+ KafkaInputDescriptor<PageView> pageViewInputDescriptor =
+ kafkaSystemDescriptor.getInputDescriptor(INPUT_PAGEVIEW_STREAM_ID, new JsonSerdeV2<>(PageView.class));
+
+ // Define a system descriptor for Azure Blob Storage
+ GenericSystemDescriptor azureBlobSystemDescriptor =
+ new GenericSystemDescriptor(OUTPUT_SYSTEM, "org.apache.samza.system.azureblob.AzureBlobSystemFactory");
+
+ GenericOutputDescriptor<PageViewAvroRecord> azureBlobOuputDescriptor =
+ azureBlobSystemDescriptor.getOutputDescriptor(OUTPUT_STREAM, new NoOpSerde<>());
+
+ // Set Kafka as the default system for the job
+ appDescriptor.withDefaultSystem(kafkaSystemDescriptor);
+
+ // Define the input and output streams with descriptors
+ MessageStream<PageView> pageViewInput = appDescriptor.getInputStream(pageViewInputDescriptor);
+ OutputStream<PageViewAvroRecord> pageViewAvroRecordOutputStream = appDescriptor.getOutputStream(azureBlobOuputDescriptor);
+
+ // Define the execution flow with the high-level API
+ pageViewInput
+ .map((message) -> {
+ System.out.println("Sending: ");
+ System.out.println("Received PageViewEvent with pageId: " + message.pageId);
Review comment:
Consider using a logger for this.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371585166
##########
File path: src/main/config/azure-blob-application.properties
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Job
+job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
+job.name=azure-blob
+
+# YARN package path
+yarn.package.path=file://${basedir}/target/${project.artifactId}-${pom.version}-dist.tar.gz
+
+# StreamApplication class
+app.class=samza.examples.azure.AzureBlobApplication
+
+#Azure blob essential configs
+systems.oss-testcontainer.samza.factory=org.apache.samza.system.azureblob.AzureBlobSystemFactory
+sensitive.systems.oss-testcontainer.azureblob.account.name=your-azure-storage-account-name
+sensitive.systems.oss-testcontainer.azureblob.account.key=your-azure-storage-account-key
+
+#Azure blob config - to created a blob per 2 input kafka messages
+systems.oss-testcontainer.azureblob.maxMessagesPerBlob=2
Review comment:
"oss-testcontainer" is not really a descriptive term here. Can you just use "azure-blob" (or something like that)? If it needs to be a specific format, then please explain that in the comments for this file.
It looks like you put instructions in `AzureBlobApplication`. Maybe put a note in this file that there are usage instructions in that class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] lakshmi-manasa-g commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
lakshmi-manasa-g commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371926907
##########
File path: src/main/java/samza/examples/azure/AzureBlobApplication.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package samza.examples.azure;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+import org.apache.samza.operators.MessageStream;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.serializers.JsonSerdeV2;
+import org.apache.samza.serializers.NoOpSerde;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
+import samza.examples.azure.data.PageViewAvroRecord;
+import samza.examples.cookbook.data.PageView;
+
+/**
+ * In this example, we demonstrate sending blobs to Azure Blob Storage.
+ * This Samza job reads from Kafka topic "page-view-azure-blob-input" and produces blobs to Azure-Container "oss-testcontainer" in your Azure Storage account.
+ *
+ * Currently, Samza supports sending Avro files are blobs.
+ * Hence the incoming messages into the Samza job have to be converted to an Avro record.
+ * For this job, we use input message as {@link samza.examples.cookbook.data.PageView} and
+ * covert it to an Avro record defined as {@link samza.examples.azure.data.PageViewAvroRecord}.
+ *
+ * To run the below example:
+ *
+ * <ol>
+ * <li>
+ * Replace your-azure-storage-account-name and your-azure-storage-account-key with details of your Azure Storage Account.
+ * </li>
+ * <li>
+ * Ensure that the topic "page-view-azure-blob-input" is created <br/>
+ * ./deploy/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic page-view-azure-blob-input --partitions 1 --replication-factor 1
+ * </li>
+ * <li>
+ * Run the application using the run-app.sh script <br/>
+ * ./deploy/samza/bin/run-app.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/azure-blob-application.properties
+ * </li>
+ * <li>
+ * Produce some messages to the "page-view-azure-blob-input" topic <br/>
+ * ./deploy/kafka/bin/kafka-console-producer.sh --topic page-view-azure-blob-input --broker-list localhost:9092 <br/>
+ * {"userId": "user1", "country": "india", "pageId":"google.com"} <br/>
+ * {"userId": "user2", "country": "france", "pageId":"facebook.com"} <br/>
+ * {"userId": "user3", "country": "china", "pageId":"yahoo.com"} <br/>
+ * {"userId": "user4", "country": "italy", "pageId":"linkedin.com"} <br/>
+ * {"userId": "user5", "country": "germany", "pageId":"amazon.com"} <br/>
+ * {"userId": "user6", "country": "denmark", "pageId":"apple.com"} <br/>
+ * </li>
+ * <li>
+ * Seeing Output:
+ * <ol>
+ * <li>
+ * See blobs in your Azure portal at https://<azure-storage-account-name>.blob.core.windows.net/oss-testcontainer/PageViewEventStream/<time-stamp>.avro
+ * </li>
+ * <li>
+ * system-name "oss-testcontainer" in configs and code below maps to Azure-Container in Azure Storage account.
+ * </li>
+ * <li>
+ * <time-stamp> is of the format yyyy/MM/dd/HH/mm-ss-randomString.avro. Hence navigate through the virtual folders on the portal to see your blobs.
+ * </li>
+ * <li>
+ * Due to network calls, allow a few minutes for blobs to appear on the portal.
+ * </li>
+ * <li>
+ * Config "maxMessagesPerBlob=2" ensures that a blob is created per 2 input messages. Adjust input or config accordingly.
+ * </li>
+ * </ol>
+ * </li>
+ * </ol>
+ */
+public class AzureBlobApplication implements StreamApplication {
+ private static final List<String> KAFKA_CONSUMER_ZK_CONNECT = ImmutableList.of("localhost:2181");
+ private static final List<String> KAFKA_PRODUCER_BOOTSTRAP_SERVERS = ImmutableList.of("localhost:9092");
+ private static final Map<String, String> KAFKA_DEFAULT_STREAM_CONFIGS = ImmutableMap.of("replication.factor", "1");
+ private static final String INPUT_PAGEVIEW_STREAM_ID = "page-view-azure-blob-input";
+ private static final String OUTPUT_SYSTEM = "oss-testcontainer";
+ private static final String OUTPUT_STREAM = "PageViewEventStream";
+
+ @Override
+ public void describe(StreamApplicationDescriptor appDescriptor) {
+ // Define a system descriptor for Kafka
+ KafkaSystemDescriptor kafkaSystemDescriptor =
+ new KafkaSystemDescriptor("kafka").withConsumerZkConnect(KAFKA_CONSUMER_ZK_CONNECT)
+ .withProducerBootstrapServers(KAFKA_PRODUCER_BOOTSTRAP_SERVERS)
+ .withDefaultStreamConfigs(KAFKA_DEFAULT_STREAM_CONFIGS);
+
+ KafkaInputDescriptor<PageView> pageViewInputDescriptor =
+ kafkaSystemDescriptor.getInputDescriptor(INPUT_PAGEVIEW_STREAM_ID, new JsonSerdeV2<>(PageView.class));
+
+ // Define a system descriptor for Azure Blob Storage
+ GenericSystemDescriptor azureBlobSystemDescriptor =
+ new GenericSystemDescriptor(OUTPUT_SYSTEM, "org.apache.samza.system.azureblob.AzureBlobSystemFactory");
Review comment:
I did write a system descriptor for Azure Blob but then it was not a lot more than the generic system descriptor and hence decided to use the existing generic system descriptor itself.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371968280
##########
File path: pom.xml
##########
@@ -206,6 +206,26 @@ under the License.
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
+ <!-- Jackson Dependencies -->
Review comment:
Does that mean that the azure library needs Jackson, but it does not pull it in transitively? That seems odd. Or maybe check if some other dependency is using a version of Jackson that is incompatible with the azure client.
In any case, can you please add a comment about why this is necessary? It might be something you need to note on a user guide if there is one.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371584567
##########
File path: pom.xml
##########
@@ -206,6 +206,26 @@ under the License.
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
+ <!-- Jackson Dependencies -->
Review comment:
Does the app depend directly on Jackson? If not, then you shouldn't need these dependencies.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371586933
##########
File path: src/main/java/samza/examples/azure/AzureBlobApplication.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package samza.examples.azure;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+import org.apache.samza.operators.MessageStream;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.serializers.JsonSerdeV2;
+import org.apache.samza.serializers.NoOpSerde;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
+import samza.examples.azure.data.PageViewAvroRecord;
+import samza.examples.cookbook.data.PageView;
+
+/**
+ * In this example, we demonstrate sending blobs to Azure Blob Storage.
+ * This Samza job reads from Kafka topic "page-view-azure-blob-input" and produces blobs to Azure-Container "oss-testcontainer" in your Azure Storage account.
+ *
+ * Currently, Samza supports sending Avro files are blobs.
+ * Hence the incoming messages into the Samza job have to be converted to an Avro record.
+ * For this job, we use input message as {@link samza.examples.cookbook.data.PageView} and
+ * covert it to an Avro record defined as {@link samza.examples.azure.data.PageViewAvroRecord}.
+ *
+ * To run the below example:
+ *
+ * <ol>
+ * <li>
+ * Replace your-azure-storage-account-name and your-azure-storage-account-key with details of your Azure Storage Account.
+ * </li>
+ * <li>
+ * Ensure that the topic "page-view-azure-blob-input" is created <br/>
+ * ./deploy/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic page-view-azure-blob-input --partitions 1 --replication-factor 1
+ * </li>
+ * <li>
+ * Run the application using the run-app.sh script <br/>
+ * ./deploy/samza/bin/run-app.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/azure-blob-application.properties
+ * </li>
+ * <li>
+ * Produce some messages to the "page-view-azure-blob-input" topic <br/>
+ * ./deploy/kafka/bin/kafka-console-producer.sh --topic page-view-azure-blob-input --broker-list localhost:9092 <br/>
+ * {"userId": "user1", "country": "india", "pageId":"google.com"} <br/>
+ * {"userId": "user2", "country": "france", "pageId":"facebook.com"} <br/>
+ * {"userId": "user3", "country": "china", "pageId":"yahoo.com"} <br/>
+ * {"userId": "user4", "country": "italy", "pageId":"linkedin.com"} <br/>
+ * {"userId": "user5", "country": "germany", "pageId":"amazon.com"} <br/>
+ * {"userId": "user6", "country": "denmark", "pageId":"apple.com"} <br/>
+ * </li>
+ * <li>
+ * Seeing Output:
+ * <ol>
+ * <li>
+ * See blobs in your Azure portal at https://<azure-storage-account-name>.blob.core.windows.net/oss-testcontainer/PageViewEventStream/<time-stamp>.avro
+ * </li>
+ * <li>
+ * system-name "oss-testcontainer" in configs and code below maps to Azure-Container in Azure Storage account.
+ * </li>
+ * <li>
+ * <time-stamp> is of the format yyyy/MM/dd/HH/mm-ss-randomString.avro. Hence navigate through the virtual folders on the portal to see your blobs.
+ * </li>
+ * <li>
+ * Due to network calls, allow a few minutes for blobs to appear on the portal.
+ * </li>
+ * <li>
+ * Config "maxMessagesPerBlob=2" ensures that a blob is created per 2 input messages. Adjust input or config accordingly.
+ * </li>
+ * </ol>
+ * </li>
+ * </ol>
+ */
+public class AzureBlobApplication implements StreamApplication {
+ private static final List<String> KAFKA_CONSUMER_ZK_CONNECT = ImmutableList.of("localhost:2181");
+ private static final List<String> KAFKA_PRODUCER_BOOTSTRAP_SERVERS = ImmutableList.of("localhost:9092");
+ private static final Map<String, String> KAFKA_DEFAULT_STREAM_CONFIGS = ImmutableMap.of("replication.factor", "1");
+ private static final String INPUT_PAGEVIEW_STREAM_ID = "page-view-azure-blob-input";
+ private static final String OUTPUT_SYSTEM = "oss-testcontainer";
+ private static final String OUTPUT_STREAM = "PageViewEventStream";
+
+ @Override
+ public void describe(StreamApplicationDescriptor appDescriptor) {
+ // Define a system descriptor for Kafka
+ KafkaSystemDescriptor kafkaSystemDescriptor =
+ new KafkaSystemDescriptor("kafka").withConsumerZkConnect(KAFKA_CONSUMER_ZK_CONNECT)
+ .withProducerBootstrapServers(KAFKA_PRODUCER_BOOTSTRAP_SERVERS)
+ .withDefaultStreamConfigs(KAFKA_DEFAULT_STREAM_CONFIGS);
+
+ KafkaInputDescriptor<PageView> pageViewInputDescriptor =
+ kafkaSystemDescriptor.getInputDescriptor(INPUT_PAGEVIEW_STREAM_ID, new JsonSerdeV2<>(PageView.class));
+
+ // Define a system descriptor for Azure Blob Storage
+ GenericSystemDescriptor azureBlobSystemDescriptor =
+ new GenericSystemDescriptor(OUTPUT_SYSTEM, "org.apache.samza.system.azureblob.AzureBlobSystemFactory");
Review comment:
It might be nice to have a system descriptor for the azure blob system, but probably not necessary now. The benefit is that you could move some of the properties into this file directly.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371968359
##########
File path: pom.xml
##########
@@ -206,6 +206,26 @@ under the License.
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
+ <!-- Jackson Dependencies -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>2.10.0</version>
+ </dependency>
+
Review comment:
minor: whitespace
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371586679
##########
File path: src/main/java/samza/examples/azure/data/PageViewAvroRecord.java
##########
@@ -0,0 +1,48 @@
+package samza.examples.azure.data;
+
+import java.io.Serializable;
+import org.apache.avro.AvroRuntimeException;
+import samza.examples.cookbook.data.PageView;
+
+public class PageViewAvroRecord extends org.apache.avro.specific.SpecificRecordBase
+ implements org.apache.avro.specific.SpecificRecord, Serializable {
+ public final org.apache.avro.Schema SCHEMA = org.apache.avro.Schema.parse(
+ "{\"type\":\"record\",\"name\":\"PageViewAvroRecord\",\"namespace\":\"org.apache.samza.examples.events\", \"fields\":[{\"name\": \"userId\", \"type\": \"string\"}, {\"name\": \"country\", \"type\": \"string\"}, {\"name\": \"pageId\", \"type\": \"string\"}]}");
+
+ private String userId;
+ private String country;
+ private String pageId;
+
+ public static PageViewAvroRecord buildPageViewRecord(PageView pageView) {
+ PageViewAvroRecord record = new PageViewAvroRecord();
+ record.put(0, pageView.userId);
Review comment:
Would it be cleaner to do `record.userId = pageView.userId`? Same for the other fields below.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] lakshmi-manasa-g commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
lakshmi-manasa-g commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r379603189
##########
File path: pom.xml
##########
@@ -206,6 +206,26 @@ under the License.
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
+ <!-- Jackson Dependencies -->
Review comment:
updating jackson-core to 2.10.0 in Samza-azure (https://github.com/apache/samza/pull/1277)
after pulling in recent commits from 'latest' branch and updating jackson-core in samza-azure, these dependencies are not needed here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 merged pull request #71:
SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 merged pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371968434
##########
File path: pom.xml
##########
@@ -206,6 +206,26 @@ under the License.
<artifactId>guava</artifactId>
<version>23.0</version>
</dependency>
+ <!-- Jackson Dependencies -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>2.10.0</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.10.0</version>
+ </dependency>
+
+ <dependency>
+
Review comment:
minor: whitespace
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [samza-hello-samza] cameronlee314 commented on a change in pull
request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
Posted by GitBox <gi...@apache.org>.
cameronlee314 commented on a change in pull request #71: SAMZA-2437: Sample for producing to Azure Blob Storage
URL: https://github.com/apache/samza-hello-samza/pull/71#discussion_r371585315
##########
File path: src/main/config/azure-blob-application.properties
##########
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Job
+job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
+job.name=azure-blob
+
+# YARN package path
+yarn.package.path=file://${basedir}/target/${project.artifactId}-${pom.version}-dist.tar.gz
+
+# StreamApplication class
+app.class=samza.examples.azure.AzureBlobApplication
+
+#Azure blob essential configs
+systems.oss-testcontainer.samza.factory=org.apache.samza.system.azureblob.AzureBlobSystemFactory
+sensitive.systems.oss-testcontainer.azureblob.account.name=your-azure-storage-account-name
+sensitive.systems.oss-testcontainer.azureblob.account.key=your-azure-storage-account-key
+
+#Azure blob config - to created a blob per 2 input kafka messages
+systems.oss-testcontainer.azureblob.maxMessagesPerBlob=2
+
+# Add configuration to disable checkpointing for this job once it is available in the Coordinator Stream model
+# See https://issues.apache.org/jira/browse/SAMZA-465?focusedCommentId=14533346&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14533346 for more details
Review comment:
Is this necessary?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services