Hi can someone help me with writing data to an event hub with spark streaming? I've been following the tutorial here: https://github.com/Azure/azure-event-hubs-spark/blob/master/docs/structured-streaming-eventhubs-integration.md#creating-an-eventhubs-sink-for-streaming-queries. This is the code I have for the final step to write to event hub.
df
.select("body")
.writeStream
.format("eventhubs")
.options(readyToCookEventhubParameters.toMap)
.option("checkpointLocation", "abfss://checkpointpathnotincluded")
.start()
.awaitTermination()
I know the dataframe has the correct data inside, but was wondering if anyone had ideas on what i might be missing to prevent the write from completing? I'm checking if the writes are done with the Process data tab in the portal.
I don't get any error messages but I did notice the sink information is very sparse:
"sink" : {
"description" : "EventHubsSink"
}
Thanks!
val eventHubStream = EventHubsUtils.createDirectStream(ssc, ehConf)
val FiveMiniteSlidingWindoStream = eventHubStream.window(Minutes(2), Minutes(1))
eventHubStream.foreachRDD { eventHubEvents =>
eventHubEvents.foreach { r => {
println(s"Event bytes inside for Before fcn are ${r.getBytes}")
}
}
Helper.processEventHubEvents(eventHubEvents, docDb.value, 1)
}
FiveMiniteSlidingWindoStream.foreachRDD { eventHubEvents =>
Helper.processEventHubEvents(eventHubEvents, docDb.value,5)
} 
Are there any plans on integrating this authentication method in the connector? How is the best way to raise this feature request?
[INFO] Cannot access type Experimental in package org.apache.http.annotation. The current classpath may be
[INFO] missing a definition for org.apache.http.annotation.Experimental, or EventHubsConf.class may have been compiled against a version that's
[INFO] incompatible with the one found on the current classpath.
[ERROR] one error found
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<spark.version>2.4.3</spark.version>
<scalatest.version>3.2.0-M1</scalatest.version>
<scala.version>2.11</scala.version>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_${scala.version}</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>com.microsoft.azure</groupId>
<artifactId>azure-eventhubs-spark_${scala.version}</artifactId>
<version>2.3.10</version>
</dependency>
<dependency>
<groupId>com.microsoft.azure</groupId>
<artifactId>azure-eventhubs</artifactId>
<version>2.3.0</version>
</dependency>
I'm using Spark 2.4, with Event Hubs 2.3.14.1
Based on this documentation
https://github.com/Azure/azure-event-hubs-spark/blob/master/docs/structured-streaming-eventhubs-integration.md I can create a writeStream like so
val query =
streamingSelectDF
.writeStream
.foreach(writer)
.outputMode("update")
.trigger(ProcessingTime("25 seconds"))
.start()Error:(42, 16) type mismatch;
found : org.apache.spark.sql.eventhubs.EventHubsForeachWriter
required: org.apache.spark.sql.ForeachWriter[org.apache.spark.sql.Row]
.foreach(writer)Hello again.
I'm trying to use Event Hubs connector to write JSON messages into the file system. It should be parquet format, but I'm just playing around right now
val ehConf = EventHubsConf(connectionString)
.setConsumerGroup("audit_spark_stream")
.setStartingPosition(EventPosition.fromStartOfStream)
val df = spark
.readStream
.format("eventhubs")
.options(ehConf.toMap).load()
df.selectExpr( "CAST(body AS STRING)")
val query = df.writeStream
.format("json")
.option("path","data/audit/json")
.option("checkpointLocation","data/audit/jsoncp")
.trigger(ProcessingTime("25 seconds"))
.start().awaitTermination()The stpark job compiles and starts fine, but it's not writing data out. Does anyone see anything wrong with the approach?
val ehConf = EventHubsConf(connectionString)
.setConsumerGroup("audit_spark_stream")
.setStartingPosition(EventPosition.fromSequenceNumber(0L))
.setEndingPosition(EventPosition.fromEnqueuedTime(Instant.now))
// Simple batch query
val df = spark.read
.format("eventhubs")
.options(ehConf.toMap)
.load()
import spark.implicits._
df.select($"body" cast "string").show()
val positions = Map(
new NameAndPartition(name, 0) -> EventPosition.fromStartOfStream,
new NameAndPartition(name, 1) -> EventPosition.fromSequenceNumber(100L)
)
val cs = "YOUR.CONNECTION.STRING"
val ehConf = EventHubsConf(cs)
.setStartingPositions(positions)
.setStartingPosition(EventPosition.fromEndOfStream)
Streaming is start from @Latest for all the partition ...
Even if i remove .setStartingPosition(EventPosition.fromEndOfStream) also , It takes from @Latest position ...
return self._df(self._jreader.load())
File "/opt/anaconda3/envs/bidh-adls/lib/python3.7/site-packages/py4j/java_gateway.py", line 1257, in call
answer, self.gateway_client, self.target_id, self.name)
File "/opt/anaconda3/envs/bidh-adls/lib/python3.7/site-packages/pyspark/sql/utils.py", line 63, in deco
return f(a, *kw)
File "/opt/anaconda3/envs/bidh-adls/lib/python3.7/site-packages/py4j/protocol.py", line 328, in get_return_value
format(target_id, ".", name), value)
py4j.protocol.Py4JJavaError: An error occurred while calling o24.load.
: java.lang.ClassNotFoundException: Failed to find data source: org.apache.spark.sql.eventhubs.EventHubsSourceProvider. Please find packages at http://spark.apache.org/third-party-projects.html
at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:657)
