DB-12042 JMS Input Source #32
Conversation
jpanko1
requested review from
arnaud-lacurie,
ascend1,
bklo94,
carolp-503,
dgomezferro,
hatyo,
ipraznik-splice,
jyuanca,
martinrupp,
msirek,
OlegMazurov and
yxia92
as code owners
| @@ -1,6 +1,6 @@ | |||
| name := "splice-machine-spark-connector" | |||
|
|
|||
| val spliceVersion = "3.2.0.2001-SNAPSHOT" | |||
| val spliceVersion = "3.1.0.2016" | |||
There was a problem hiding this comment.
Looks suspicious. A downgrade?
There was a problem hiding this comment.
It was changed to get in sync with the version of the Splice DB in the environment where this would be running.
spark3.0/build.sbt
Outdated
| // Added later separately | ||
| ExclusionRule(organization = "com.splicemachine", name = "scala_util"), | ||
| ExclusionRule(organization = "javax.ws.rs", name = "javax.ws.rs-api"), | ||
| ExclusionRule(organization = "org.apache.kafka", name = "kafka_2.11"), | ||
| ExclusionRule(organization = "org.scala-lang.modules", name = "scala-parser-combinators_2.11") | ||
| ) | ||
|
|
||
| val excludedDeps = excludedDepsNonSpark ++ Seq( |
There was a problem hiding this comment.
The following is going to be a bit faster (prepend to a head) and shorter code-wise?
val excludedDeps =
ExclusionRule(organization = "org.apache.spark") +: excludedDepsNonSpark
There was a problem hiding this comment.
Updated in commit b79cb27
| } | ||
| } | ||
| import org.apache.spark.unsafe.types.UTF8String._ | ||
| val internalRDD = messageList.map(message => InternalRow( |
There was a problem hiding this comment.
The code looks very old(ish). InternalRow conversion is not needed (as it's in in-process memory anyway). Just convert JmsMessage to whatever tuple you want and simply Seq(...).toDF(...).
|
|
||
| val query = stream.writeStream | ||
| .outputMode("append") | ||
| .format("console") |
There was a problem hiding this comment.
memory format would help you with automated testing.
| */ | ||
| class JmsSourceRdd(sc:SparkContext) extends RDD[Message](sc, Nil){ | ||
|
|
||
| override def compute(split: Partition, context: TaskContext): Iterator[Message] = ??? |
There was a problem hiding this comment.
Is this class ever used given these ????
| /** | ||
| * Created by exa00015 on 26/12/18. | ||
| */ | ||
| case class JmsSourceOffset(val id:Long) extends Offset { |
There was a problem hiding this comment.
There's a LongOffset in Spark Structured Streaming already.
|
|
||
|
|
||
| override def schema: StructType = { | ||
| ScalaReflection.schemaFor[JmsMessage].dataType.asInstanceOf[StructType] |
There was a problem hiding this comment.
What a trick! I think Encoders.product[JmsMessage].schema could work. If so, use it below to create a DataFrame out of JmsMessages.
| */ | ||
| class JmsDatasourceRelation(override val sqlContext: SQLContext, parameters: Map[String, String]) extends BaseRelation with TableScan with Serializable { | ||
|
|
||
| lazy val RECIEVER_TIMEOUT = parameters.getOrElse("reciever.timeout","3000").toLong |
| case "amq" => new AMQConnectionFactoryProvider().createConnection(parameters) | ||
| case "ibmmq" => new IBMMQConnectionFactoryProvider().createConnection(parameters) | ||
| case "rmq" => new RMQConnectionFactoryProvider().createConnection(parameters) | ||
| case "kafka" => new KafkaConnectionFactoryProvider().createConnection(parameters) |
There was a problem hiding this comment.
We don't need kafka as it's among the built-in data sources.
|
see you! |
Short Description
Support JMS input for spark structured streaming.
Long Description
This code was added to be able to read data from IBM MQ via JMS.
How to test
Bob had a server set up and generating data that the streaming code could read.