Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Allow users to define additional Spark options for SQL and Kafka sources #195

Merged
merged 1 commit into from
Jul 3, 2024
Merged

Allow users to define additional Spark options for SQL and Kafka sources #195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions tofhir-engine/src/main/scala/io/tofhir/engine/data/read/KafkaSourceReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,7 @@ class KafkaSourceReader(spark: SparkSession) extends BaseDataSourceReader[KafkaS
.option("subscribe", mappingSource.topicName)
.option("startingOffsets", mappingSource.startingOffsets)
.option("inferSchema", value = true)
.options(mappingSource.options)
.load()
.select($"value".cast(StringType)) // change the type of message from binary to string
.withColumn("value", processDataUDF(col("value"))) // replace 'value' column with the processed data
Expand Down
1 change: 1 addition & 0 deletions tofhir-engine/src/main/scala/io/tofhir/engine/data/read/SqlSourceReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ class SqlSourceReader(spark: SparkSession) extends BaseDataSourceReader[SqlSourc
.option("dbtable", dbTable)
.option("user", sourceSettings.username)
.option("password", sourceSettings.password)
.options(mappingSource.options)
.load()
}
}
6 changes: 4 additions & 2 deletions tofhir-engine/src/main/scala/io/tofhir/engine/model/FhirMappingTask.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,17 +96,19 @@ case class FileSystemSource(path: String, fileFormat:Option[String] = None, opti
*
* @param tableName Name of the table
* @param query Query to execute in the database
* @param options Further options for SQL source (Spark SQL Guide -> https://spark.apache.org/docs/3.4.1/sql-data-sources-jdbc.html ).
*/
case class SqlSource(tableName: Option[String] = None, query: Option[String] = None, override val preprocessSql: Option[String] = None) extends FhirMappingSourceContext
case class SqlSource(tableName: Option[String] = None, query: Option[String] = None, options:Map[String, String] = Map.empty[String, String], override val preprocessSql: Option[String] = None) extends FhirMappingSourceContext

/**
* Context/configuration for one of the source of the mapping that will read the source data from a kafka as stream
*
* @param topicName The topic(s) to subscribe, may be comma seperated string list (topic1,topic2)
* @param groupId The Kafka group id to use in Kafka consumer while reading from Kafka
* @param startingOffsets The start point when a query is started
* @param options Further options for Kafka source (Spark Kafka Guide -> https://spark.apache.org/docs/3.4.1/structured-streaming-kafka-integration.html)
*/
case class KafkaSource(topicName: String, groupId: String, startingOffsets: String, override val preprocessSql: Option[String] = None) extends FhirMappingSourceContext
case class KafkaSource(topicName: String, groupId: String, startingOffsets: String, options:Map[String, String] = Map.empty[String, String], override val preprocessSql: Option[String] = None) extends FhirMappingSourceContext

/**
* Represents a mapping source context for FHIR server data.
Expand Down
Loading