You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@bahir.apache.org by bchen-talend <gi...@git.apache.org> on 2017/05/03 05:55:50 UTC
[GitHub] bahir pull request #42: [BAHIR-116] Add spark streaming connector to Google ...
GitHub user bchen-talend opened a pull request:
https://github.com/apache/bahir/pull/42
[BAHIR-116] Add spark streaming connector to Google Cloud Pub/Sub
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/Talend/bahir Pubsub
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/bahir/pull/42.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #42
----
commit 1bffa3925c93f4b9b7e5721bc828fa8ec20f2ccc
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-27T09:18:32Z
[Pubsub] Initial commit to add PubsubInputDStream
(cherry picked from commit 9f19b83)
commit 88c274c3e88bb2698e34d9a876679dc51813551a
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-28T14:10:44Z
[Pubsub] complete most runtime part and add a simple test
(cherry picked from commit a35627a)
commit a819bef05c95915338390e9b4594ddc452d6c76e
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T07:10:35Z
[Pubsub] do not expose PubsubMessage to the end user
(cherry picked from commit c84a4cd)
commit 9eac2201187fb25d780d97411578d35a6683bc75
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T07:25:05Z
Merge branch 'release/2.0.0' into release/2.1.0
commit 9120d7b39285473fbc0441bd70a37dc51a2813ce
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T07:26:48Z
[Pubsub] upgrade to spark 2.1
(cherry picked from commit c84a4cd)
commit cda7b4831c6ac20e9430ce95cdaedaf5d39e4e07
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T08:52:53Z
Merge branch 'release/2.1.0'
commit 2001fb607a3137126615a923e6d8c64db09f7e51
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T08:53:52Z
[Pubsub] upgrade pubsub to spark 2.2 snapshot
commit 4185ddcfa69b70ad9d58bd9051865d76de6e4e5b
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T10:32:23Z
[Pubsub] add comment, remove unuse code
commit 713526d3625019ef656056b52ccbdf12431b292c
Author: Chen Bin <bc...@talend.com>
Date: 2017-04-29T10:40:52Z
Merge branch 'release/2.0.0' into release/2.1.0
commit 81282cba1dd8ce295d0002ae590da2ae5ca6c61f
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-02T08:02:51Z
[Pubsub] refact how to get credentials and add document & test
commit 6501a14f8768b698f2a11aeaf73c4dd4d74e3bd2
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-02T08:04:46Z
Merge branch 'release/2.0.0' into release/2.1.0
commit cfa6c5fe2e66cf80bf0f3f08364191af20ef9a65
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-02T08:21:38Z
[Pubsub] update document for 2.1
commit ee0b1e80a97054b194e65a3392c40750f8e9e6d3
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-03T03:45:55Z
[Pubsub] add example and fix one bug
commit dc3431a407c8bff34d4c1c42e20934f342b57d59
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-03T05:28:10Z
[Pubsub] add example for credential in README.md
commit 85faaa72a8e088efee898a6c28760c3cfac41944
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-03T05:32:30Z
Merge branch 'release/2.0.0' into release/2.1.0
commit c828163fa491cc09bc66e847a6d3ca44454f21aa
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-03T05:39:07Z
Merge branch 'release/2.1.0'
commit cb274cec3e454de88db78fc47cdf765e50187fe2
Author: Chen Bin <bc...@talend.com>
Date: 2017-05-03T05:42:04Z
[Pubsub] update document for 2.2-snapshot
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Refer to this link for build results (access rights to CI server needed):
http://169.45.79.58:8080/job/bahir_spark_pr_builder/56/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir pull request #42: [BAHIR-116] Add spark streaming connector to Google ...
Posted by amarouni <gi...@git.apache.org>.
Github user amarouni commented on a diff in the pull request:
https://github.com/apache/bahir/pull/42#discussion_r114502066
--- Diff: streaming-pubsub/examples/src/main/scala/org.apache.spark.examples.streaming.pubsub/PubsubWordCount.scala ---
@@ -0,0 +1,150 @@
+/*
+ * 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.
+ */
+
+// scalastyle:off println
+package org.apache.spark.examples.streaming.pubsub
+
+import scala.collection.JavaConverters._
+import scala.util.Random
+
+import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
+import com.google.api.client.json.jackson2.JacksonFactory
+import com.google.api.services.pubsub.Pubsub.Builder
+import com.google.api.services.pubsub.model.PublishRequest
+import com.google.api.services.pubsub.model.PubsubMessage
+import com.google.cloud.hadoop.util.RetryHttpInitializer
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.pubsub.ConnectionUtils
+import org.apache.spark.streaming.pubsub.PubsubTestUtils
+import org.apache.spark.streaming.pubsub.PubsubUtils
+import org.apache.spark.streaming.pubsub.SparkGCPCredentials
+import org.apache.spark.streaming.Milliseconds
+import org.apache.spark.streaming.StreamingContext
+import org.apache.spark.SparkConf
+
+/**
+ * Consumes messages from a Google Cloud Pub/Sub subscription and does wordcount.
+ * In this example it use application default credentials, so need to use gcloud
+ * client to generate token file before running example
+ *
+ * Usage: PubsubWordCount <projectId> <subscription>
+ * <projectId> is the name of Google cloud
+ * <subscription> is the subscription to a topic
+ *
+ * Example:
+ * # use gcloud client generate token file
+ * $ gcloud init
+ * $ gcloud auth application-default login
+ *
+ * # run the example
+ * $ bin/run-example \
+ * org.apache.spark.examples.streaming.pubsub.PubsubWordCount project_1 subscription_1
+ *
+ */
+object PubsubWordCount {
+ def main(args: Array[String]): Unit = {
+ if (args.length != 2) {
+ System.err.println(
+ """
+ |Usage: PubsubWordCount <projectId> <subscription>
+ |
+ | <projectId> is the name of Google cloud
+ | <subscription> is the subscription to a topic
+ |
+ """.stripMargin)
+ System.exit(1)
+ }
+
+ val Seq(projectId, subscription) = args.toSeq
+
+ val sparkConf = new SparkConf().setAppName("PubsubWordCount")
+ val ssc = new StreamingContext(sparkConf, Milliseconds(2000))
+
+ val pubsubStream = PubsubUtils.createStream(ssc, projectId, subscription,
+ SparkGCPCredentials.builder.build(), StorageLevel.MEMORY_AND_DISK_SER_2)
+
+ val wordCounts =
+ pubsubStream.map(message => (new String(message.getData()), 1)).reduceByKey(_ + _)
+
+ wordCounts.print()
+
+ ssc.start()
+ ssc.awaitTermination()
+ }
+
+}
+
+/**
--- End diff --
@bchen-talend Can you add a little description as for the above example ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Build failed, see build log for details
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Build successful
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Can one of the admins verify this patch?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by lresende <gi...@git.apache.org>.
Github user lresende commented on the issue:
https://github.com/apache/bahir/pull/42
ok to test
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir pull request #42: [BAHIR-116] Add spark streaming connector to Google ...
Posted by amarouni <gi...@git.apache.org>.
Github user amarouni commented on a diff in the pull request:
https://github.com/apache/bahir/pull/42#discussion_r114494439
--- Diff: streaming-pubsub/src/main/scala/org/apache/spark/streaming/pubsub/PubsubInputDStream.scala ---
@@ -0,0 +1,259 @@
+/*
+ * 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 org.apache.spark.streaming.pubsub
+
+import java.io.{Externalizable, ObjectInput, ObjectOutput}
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
+
+import com.google.api.client.auth.oauth2.Credential
+import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
+import com.google.api.client.googleapis.json.GoogleJsonResponseException
+import com.google.api.client.json.jackson2.JacksonFactory
+import com.google.api.services.pubsub.Pubsub.Builder
+import com.google.api.services.pubsub.model.{AcknowledgeRequest, PubsubMessage, PullRequest}
+import com.google.cloud.hadoop.util.RetryHttpInitializer
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.StreamingContext
+import org.apache.spark.streaming.dstream.ReceiverInputDStream
+import org.apache.spark.streaming.receiver.Receiver
+import org.apache.spark.util.Utils
+
+/**
+ * Input stream that subscribe messages from Google cloud Pub/Sub subscription.
+ * @param project Google cloud project id
+ * @param subscription Pub/Sub subscription name
+ * @param credential Google cloud project credential to access Pub/Sub service
+ */
+private[streaming]
+class PubsubInputDStream(
+ _ssc: StreamingContext,
+ val project: String,
+ val subscription: String,
+ val credential: SparkGCPCredentials,
+ val _storageLevel: StorageLevel
+) extends ReceiverInputDStream[SparkPubsubMessage](_ssc) {
+
+ override def getReceiver(): Receiver[SparkPubsubMessage] = {
+ new PubsubReceiver(project, subscription, credential, _storageLevel)
+ }
+}
+
+/**
+ * A wrapper class for PubsubMessage's with a custom serialization format.
+ *
+ * This is necessary because PubsubMessage uses inner data structures
+ * which are not serializable.
+ */
+class SparkPubsubMessage() extends Externalizable {
+
+ private[pubsub] var message = new PubsubMessage
+
+ def getData(): Array[Byte] = message.decodeData()
+
+ def getAttributes(): java.util.Map[String, String] = message.getAttributes
+
+ def getMessageId(): String = message.getMessageId
+
+ def getPublishTime(): String = message.getPublishTime
+
+ override def writeExternal(out: ObjectOutput): Unit = Utils.tryOrIOException {
+ message.decodeData() match {
+ case null => out.writeInt(-1)
+ case data =>
+ out.writeInt(data.size)
+ out.write(data)
+ }
+
+ message.getMessageId match {
+ case null => out.writeInt(-1)
+ case id =>
+ val idBuff = Utils.serialize(id)
+ out.writeInt(idBuff.length)
+ out.write(idBuff)
+ }
+
+ message.getPublishTime match {
+ case null => out.writeInt(-1)
+ case time =>
+ val publishTimeBuff = Utils.serialize(time)
+ out.writeInt(publishTimeBuff.length)
+ out.write(publishTimeBuff)
+ }
+
+ message.getAttributes match {
+ case null => out.writeInt(-1)
+ case attrs =>
+ out.writeInt(attrs.size())
+ for ((k, v) <- message.getAttributes.asScala) {
+ val keyBuff = Utils.serialize(k)
+ out.writeInt(keyBuff.length)
+ out.write(keyBuff)
+ val valBuff = Utils.serialize(v)
+ out.writeInt(valBuff.length)
+ out.write(valBuff)
+ }
+ }
+ }
+
+ override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException {
+ in.readInt() match {
+ case -1 => message.encodeData(null)
+ case bodyLength =>
+ val data = new Array[Byte](bodyLength)
+ in.readFully(data)
+ message.encodeData(data)
+ }
+
+ in.readInt() match {
+ case -1 => message.setMessageId(null)
+ case idLength =>
+ val idBuff = new Array[Byte](idLength)
+ in.readFully(idBuff)
+ val id: String = Utils.deserialize(idBuff)
+ message.setMessageId(id)
+ }
+
+ in.readInt() match {
+ case -1 => message.setPublishTime(null)
+ case publishTimeLength =>
+ val publishTimeBuff = new Array[Byte](publishTimeLength)
+ in.readFully(publishTimeBuff)
+ val publishTime: String = Utils.deserialize(publishTimeBuff)
+ message.setPublishTime(publishTime)
+ }
+
+ in.readInt() match {
+ case -1 => message.setAttributes(null)
+ case numAttributes =>
+ val attributes = new java.util.HashMap[String, String]
+ for (i <- 0 until numAttributes) {
+ val keyLength = in.readInt()
+ val keyBuff = new Array[Byte](keyLength)
+ in.readFully(keyBuff)
+ val key: String = Utils.deserialize(keyBuff)
+
+ val valLength = in.readInt()
+ val valBuff = new Array[Byte](valLength)
+ in.readFully(valBuff)
+ val value: String = Utils.deserialize(valBuff)
+
+ attributes.put(key, value)
+ }
+ message.setAttributes(attributes)
+ }
+ }
+}
+
+private [pubsub]
+object ConnectionUtils {
+ val transport = GoogleNetHttpTransport.newTrustedTransport();
+ val jacksonFactory = JacksonFactory.getDefaultInstance;
+
+ /**
+ * Client can retry with these response status
+ */
+ val RESOURCE_EXHAUSTED = 429
+
+ val CANCELLED = 499
+
+ val INTERNAL = 500
+
+ val UNAVAILABLE = 503
+
+ val DEADLINE_EXCEEDED = 504
+
+ def retryable(status: Int): Boolean = {
+ status match {
+ case RESOURCE_EXHAUSTED | CANCELLED | INTERNAL | UNAVAILABLE | DEADLINE_EXCEEDED => true
+ case _ => false
+ }
+ }
+}
+
+
+private[pubsub]
+class PubsubReceiver(
+ project: String,
+ subscription: String,
+ credential: SparkGCPCredentials,
+ storageLevel: StorageLevel)
+ extends Receiver[SparkPubsubMessage](storageLevel) {
+
+ val APP_NAME = "sparkstreaming-pubsub-receiver"
+
+ val INIT_BACKOFF = 100 // 100ms
--- End diff --
@bchen-talend Can this value (and the other values beneath it) be supplied by the user ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by lresende <gi...@git.apache.org>.
Github user lresende commented on the issue:
https://github.com/apache/bahir/pull/42
@bchen-talend Sorry for delaying the review, I am at a conference and finishing up slides/demo and should be able to get to this after today.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir pull request #42: [BAHIR-116] Add spark streaming connector to Google ...
Posted by amarouni <gi...@git.apache.org>.
Github user amarouni commented on a diff in the pull request:
https://github.com/apache/bahir/pull/42#discussion_r114503784
--- Diff: streaming-pubsub/examples/src/main/scala/org.apache.spark.examples.streaming.pubsub/PubsubWordCount.scala ---
@@ -0,0 +1,150 @@
+/*
+ * 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.
+ */
+
+// scalastyle:off println
+package org.apache.spark.examples.streaming.pubsub
+
+import scala.collection.JavaConverters._
+import scala.util.Random
+
+import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
+import com.google.api.client.json.jackson2.JacksonFactory
+import com.google.api.services.pubsub.Pubsub.Builder
+import com.google.api.services.pubsub.model.PublishRequest
+import com.google.api.services.pubsub.model.PubsubMessage
+import com.google.cloud.hadoop.util.RetryHttpInitializer
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.pubsub.ConnectionUtils
+import org.apache.spark.streaming.pubsub.PubsubTestUtils
+import org.apache.spark.streaming.pubsub.PubsubUtils
+import org.apache.spark.streaming.pubsub.SparkGCPCredentials
+import org.apache.spark.streaming.Milliseconds
+import org.apache.spark.streaming.StreamingContext
+import org.apache.spark.SparkConf
+
+/**
+ * Consumes messages from a Google Cloud Pub/Sub subscription and does wordcount.
+ * In this example it use application default credentials, so need to use gcloud
+ * client to generate token file before running example
+ *
+ * Usage: PubsubWordCount <projectId> <subscription>
+ * <projectId> is the name of Google cloud
+ * <subscription> is the subscription to a topic
+ *
+ * Example:
+ * # use gcloud client generate token file
+ * $ gcloud init
+ * $ gcloud auth application-default login
+ *
+ * # run the example
+ * $ bin/run-example \
+ * org.apache.spark.examples.streaming.pubsub.PubsubWordCount project_1 subscription_1
+ *
+ */
+object PubsubWordCount {
+ def main(args: Array[String]): Unit = {
+ if (args.length != 2) {
+ System.err.println(
+ """
+ |Usage: PubsubWordCount <projectId> <subscription>
+ |
+ | <projectId> is the name of Google cloud
+ | <subscription> is the subscription to a topic
+ |
+ """.stripMargin)
+ System.exit(1)
+ }
+
+ val Seq(projectId, subscription) = args.toSeq
+
+ val sparkConf = new SparkConf().setAppName("PubsubWordCount")
+ val ssc = new StreamingContext(sparkConf, Milliseconds(2000))
+
+ val pubsubStream = PubsubUtils.createStream(ssc, projectId, subscription,
--- End diff --
It might be more helpful to type this variable, this gives the users any idea of the DStream type. As most users will use this example as the starting point for using the connector.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by bchen-talend <gi...@git.apache.org>.
Github user bchen-talend commented on the issue:
https://github.com/apache/bahir/pull/42
@lresende No problem, when you have time, thanks :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Build successful
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir pull request #42: [BAHIR-116] Add spark streaming connector to Google ...
Posted by bchen-talend <gi...@git.apache.org>.
Github user bchen-talend closed the pull request at:
https://github.com/apache/bahir/pull/42
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir pull request #42: [BAHIR-116] Add spark streaming connector to Google ...
Posted by bchen-talend <gi...@git.apache.org>.
Github user bchen-talend commented on a diff in the pull request:
https://github.com/apache/bahir/pull/42#discussion_r114743438
--- Diff: streaming-pubsub/examples/src/main/scala/org.apache.spark.examples.streaming.pubsub/PubsubWordCount.scala ---
@@ -0,0 +1,150 @@
+/*
+ * 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.
+ */
+
+// scalastyle:off println
+package org.apache.spark.examples.streaming.pubsub
+
+import scala.collection.JavaConverters._
+import scala.util.Random
+
+import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport
+import com.google.api.client.json.jackson2.JacksonFactory
+import com.google.api.services.pubsub.Pubsub.Builder
+import com.google.api.services.pubsub.model.PublishRequest
+import com.google.api.services.pubsub.model.PubsubMessage
+import com.google.cloud.hadoop.util.RetryHttpInitializer
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.pubsub.ConnectionUtils
+import org.apache.spark.streaming.pubsub.PubsubTestUtils
+import org.apache.spark.streaming.pubsub.PubsubUtils
+import org.apache.spark.streaming.pubsub.SparkGCPCredentials
+import org.apache.spark.streaming.Milliseconds
+import org.apache.spark.streaming.StreamingContext
+import org.apache.spark.SparkConf
+
+/**
+ * Consumes messages from a Google Cloud Pub/Sub subscription and does wordcount.
+ * In this example it use application default credentials, so need to use gcloud
+ * client to generate token file before running example
+ *
+ * Usage: PubsubWordCount <projectId> <subscription>
+ * <projectId> is the name of Google cloud
+ * <subscription> is the subscription to a topic
+ *
+ * Example:
+ * # use gcloud client generate token file
+ * $ gcloud init
+ * $ gcloud auth application-default login
+ *
+ * # run the example
+ * $ bin/run-example \
+ * org.apache.spark.examples.streaming.pubsub.PubsubWordCount project_1 subscription_1
+ *
+ */
+object PubsubWordCount {
+ def main(args: Array[String]): Unit = {
+ if (args.length != 2) {
+ System.err.println(
+ """
+ |Usage: PubsubWordCount <projectId> <subscription>
+ |
+ | <projectId> is the name of Google cloud
+ | <subscription> is the subscription to a topic
+ |
+ """.stripMargin)
+ System.exit(1)
+ }
+
+ val Seq(projectId, subscription) = args.toSeq
+
+ val sparkConf = new SparkConf().setAppName("PubsubWordCount")
+ val ssc = new StreamingContext(sparkConf, Milliseconds(2000))
+
+ val pubsubStream = PubsubUtils.createStream(ssc, projectId, subscription,
+ SparkGCPCredentials.builder.build(), StorageLevel.MEMORY_AND_DISK_SER_2)
+
+ val wordCounts =
+ pubsubStream.map(message => (new String(message.getData()), 1)).reduceByKey(_ + _)
+
+ wordCounts.print()
+
+ ssc.start()
+ ssc.awaitTermination()
+ }
+
+}
+
+/**
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Refer to this link for build results (access rights to CI server needed):
http://169.45.79.58:8080/job/bahir_spark_pr_builder/62/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by bchen-talend <gi...@git.apache.org>.
Github user bchen-talend commented on the issue:
https://github.com/apache/bahir/pull/42
Sorry, one more bugfix, please review, thanks
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by lresende <gi...@git.apache.org>.
Github user lresende commented on the issue:
https://github.com/apache/bahir/pull/42
@bchen-talend Thanks for the PR, I have merged it, but looks like I had a typo and it did not get closed. Could you please close this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir issue #42: [BAHIR-116] Add spark streaming connector to Google Cloud P...
Posted by ApacheBahir <gi...@git.apache.org>.
Github user ApacheBahir commented on the issue:
https://github.com/apache/bahir/pull/42
Refer to this link for build results (access rights to CI server needed):
http://169.45.79.58:8080/job/bahir_spark_pr_builder/61/
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---