[GitHub] bahir pull request #21: [BAHIR-52] Update README.md formatting for source co...
Github user lresende closed the pull request at: https://github.com/apache/bahir/pull/21 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #21: [BAHIR-52] Update README.md formatting for source co...
GitHub user lresende opened a pull request: https://github.com/apache/bahir/pull/21 [BAHIR-52] Update README.md formatting for source code Update source code paragraphs to use tabs instead of ``` which is the supported way in vanilla Jekyll. You can merge this pull request into a Git repository by running: $ git pull https://github.com/lresende/bahir readme-formatting Alternatively you can review and apply these changes as the patch at: https://github.com/apache/bahir/pull/21.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 #21 commit 6494c1fd3d26851f905bde281a692eb73c8de80f Author: Luciano Resende <lrese...@apache.org> Date: 2016-08-10T19:47:07Z [BAHIR-52] Update README.md formatting for source code Update source code paragraphs to use tabs instead of ``` which is the supported way in vanilla Jekyll. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir issue #13: [BAHIR-39] Add SQL Streaming MQTT support.
Github user lresende commented on the issue: https://github.com/apache/bahir/pull/13 LGTM Merging if there is no more comments --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir issue #13: [BAHIR-39] Add SQL Streaming MQTT support.
Github user lresende commented on the issue: https://github.com/apache/bahir/pull/13 Based on the discussion at [spark-dev](https://www.mail-archive.com/dev@spark.apache.org/msg15209.html) it seems that some of the issues raised by @frreiss are current limitations of Spark Structured Streaming. Based on that, I think we are good to merge this, and address enhancements with new jira/prs and we might need to coordinate with Spark runtime changes required... --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #13: [BAHIR-39] Add SQL Streaming MQTT support.
Github user lresende commented on a diff in the pull request:
https://github.com/apache/bahir/pull/13#discussion_r73407125
--- Diff:
sql-streaming-mqtt/src/main/scala/org/apache/bahir/utils/Logging.scala ---
@@ -0,0 +1,25 @@
+/*
+ * 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.bahir.utils
+
+import org.slf4j.LoggerFactory
+
+
+trait Logging {
--- End diff --
Do we need our own definition of Logging ? Or is ok for us to use the one
on Spark (which is what the other extensions are using)...
---
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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] bahir pull request #13: [BAHIR-39] Add SQL Streaming MQTT support.
Github user lresende commented on a diff in the pull request:
https://github.com/apache/bahir/pull/13#discussion_r73405744
--- Diff:
sql-streaming-mqtt/src/main/scala/org/apache/bahir/sql/streaming/mqtt/MQTTStreamSource.scala
---
@@ -0,0 +1,191 @@
+/*
+ * 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.bahir.sql.streaming.mqtt
+
+import java.sql.Timestamp
+import java.text.SimpleDateFormat
+import java.util.Calendar
+import java.util.concurrent.CountDownLatch
+
+import scala.collection.concurrent.TrieMap
+import scala.collection.mutable.ArrayBuffer
+import scala.util.{Failure, Success, Try}
+
+import org.apache.bahir.utils.Logging
+import org.eclipse.paho.client.mqttv3._
+import org.eclipse.paho.client.mqttv3.persist.{MemoryPersistence,
MqttDefaultFilePersistence}
+
+import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.spark.sql.execution.streaming.{LongOffset, Offset,
Source}
+import org.apache.spark.sql.sources.{DataSourceRegister,
StreamSourceProvider}
+import org.apache.spark.sql.types.{StringType, StructField, StructType,
TimestampType}
+
+
+object MQTTStreamConstants {
+
+ val DATE_FORMAT = new SimpleDateFormat("-MM-dd HH:mm:ss")
+
+ val SCHEMA_DEFAULT = StructType(StructField("value", StringType)
+:: StructField("timestamp", TimestampType) :: Nil)
+}
+
+class MQTTTextStreamSource(brokerUrl: String, persistence:
MqttClientPersistence,
+topic: String, clientId: String, messageParser: Array[Byte] =>
(String, Timestamp),
+sqlContext: SQLContext) extends Source with Logging {
+
+ override def schema: StructType = MQTTStreamConstants.SCHEMA_DEFAULT
+
+ private val store = new LocalMessageStore(persistence,
sqlContext.sparkContext.getConf)
+
+ private val messages = new TrieMap[Int, (String, Timestamp)]
+
+ private val initLock = new CountDownLatch(1)
+
+ private var offset = 0
+
+ private var client: MqttClient = _
+
+ private def fetchLastProcessedOffset(): Int = {
+Try(store.maxProcessedOffset) match {
+ case Success(x) =>
+log.info(s"Recovering from last stored offset $x")
+x
+ case Failure(e) => 0
+}
+ }
+
+ initialize()
+ private def initialize(): Unit = {
+
+client = new MqttClient(brokerUrl, clientId, persistence)
+val mqttConnectOptions: MqttConnectOptions = new MqttConnectOptions()
+mqttConnectOptions.setAutomaticReconnect(true)
+// This is required to support recovery. TODO: configurable ?
+mqttConnectOptions.setCleanSession(false)
+
+val callback = new MqttCallbackExtended() {
+
+ override def messageArrived(topic_ : String, message: MqttMessage):
Unit = synchronized {
+initLock.await() // Wait for initialization to complete.
+val temp = offset + 1
+messages.put(temp, messageParser(message.getPayload))
+offset = temp
+log.trace(s"Message arrived, $topic_ $message")
+ }
+
+ override def deliveryComplete(token: IMqttDeliveryToken): Unit = {
+ }
+
+ override def connectionLost(cause: Throwable): Unit = {
+log.warn("Connection to mqtt server lost.", cause)
+ }
+
+ override def connectComplete(reconnect: Boolean, serverURI: String):
Unit = {
+log.info(s"Connect complete $serverURI. Is it a reconnect?:
$reconnect")
+ }
+}
+client.setCallback(callback)
+client.connect(mqttConnectOptions)
+client.subscribe(topic)
+// It is not possible to initialize offset without `client.connect`
+offset = fetchLastProcessedOffset()
+initLock.countDown() // Release.
+ }
+
+ /** Stop this source and free any resources it has allocated. */
+ override d
[GitHub] bahir pull request #16: [BAHIR-30] Add basic documentation for Twitter conne...
Github user lresende closed the pull request at: https://github.com/apache/bahir/pull/16 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #18: [BAHIR-28] Add basic documentation for Akka connecto...
Github user lresende closed the pull request at: https://github.com/apache/bahir/pull/18 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #17: [BAHIR-29] Add basic documentation for MQTT Connecto...
Github user lresende closed the pull request at: https://github.com/apache/bahir/pull/17 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #15: [BAHIR-31] Add basic documentation for ZeroMQ connec...
Github user lresende closed the pull request at: https://github.com/apache/bahir/pull/15 --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #16: [BAHIR-30] Add basic documentation for Twitter conne...
GitHub user lresende opened a pull request: https://github.com/apache/bahir/pull/16 [BAHIR-30] Add basic documentation for Twitter connector You can merge this pull request into a Git repository by running: $ git pull https://github.com/lresende/bahir doc-twitter Alternatively you can review and apply these changes as the patch at: https://github.com/apache/bahir/pull/16.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 #16 commit d2c84ff23eddbe50f93b884f4869a59d7929d992 Author: Luciano Resende <lrese...@apache.org> Date: 2016-08-01T16:20:20Z [BAHIR-30] Add basic documentation for Twitter 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #18: [BAHIR-28] Add basic documentation for Akka Connecto...
GitHub user lresende opened a pull request: https://github.com/apache/bahir/pull/18 [BAHIR-28] Add basic documentation for Akka Connector You can merge this pull request into a Git repository by running: $ git pull https://github.com/lresende/bahir doc-akka Alternatively you can review and apply these changes as the patch at: https://github.com/apache/bahir/pull/18.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 #18 commit eb7899b11b9e770e7a75774dccfb2110c851a267 Author: Luciano Resende <lrese...@apache.org> Date: 2016-08-01T16:17:02Z [BAHIR-28] Add basic documentation for Akka 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #15: [BAHIR-31] Add basic documentation for ZeroMQ connec...
GitHub user lresende opened a pull request: https://github.com/apache/bahir/pull/15 [BAHIR-31] Add basic documentation for ZeroMQ connector You can merge this pull request into a Git repository by running: $ git pull https://github.com/lresende/bahir doc-zeromq Alternatively you can review and apply these changes as the patch at: https://github.com/apache/bahir/pull/15.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 #15 commit 0a6f8443ce6b6f53df565e35822ab178e03e4f96 Author: Luciano Resende <lrese...@apache.org> Date: 2016-08-01T16:21:24Z [BAHIR-31] Add basic documentation for ZeroMQ 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir pull request #17: [BAHIR-29] Add basic documentation for MQTT Connecto...
GitHub user lresende opened a pull request: https://github.com/apache/bahir/pull/17 [BAHIR-29] Add basic documentation for MQTT Connector You can merge this pull request into a Git repository by running: $ git pull https://github.com/lresende/bahir doc-mqtt Alternatively you can review and apply these changes as the patch at: https://github.com/apache/bahir/pull/17.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 #17 commit 086784aa4177449c3f73380f72b057355e85015d Author: Luciano Resende <lrese...@apache.org> Date: 2016-08-01T16:18:35Z [BAHIR-29] Add basic documentation for MQTT 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir issue #14: [BAHIR-38] clean Ivy cache during Maven install phase
Github user lresende commented on the issue: https://github.com/apache/bahir/pull/14 LGTM, merging soon --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] bahir issue #9: [BAHIR-32] - Commit to test github integration
Github user lresende commented on the issue: https://github.com/apache/bahir/pull/9 Great, this seem to be working now. --- 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 infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
