apache-spark - 如何创建自定义流数据源？

Question

我有一个用于 Spark Streaming 的自定义阅读器，可以从 WebSocket 读取数据。我将尝试 Spark Structured Streaming。

如何在 Spark Structured Streaming 中创建流数据源？

Answer 1

流数据源实现 org.apache.spark.sql.execution.streaming.Source .
org.apache.spark.sql.execution.streaming.Source 的scaladoc应该为您提供足够的入门信息(只需按照类型开发可编译的 Scala 类型)。

一旦您拥有 Source您必须注册它才能在 format 中使用它的 DataStreamReader .使流媒体源可用以便您可以将其用于 format 的技巧是通过创建 DataSourceRegister 来注册它对于流媒体源。您可以在 META-INF/services/org.apache.spark.sql.sources.DataSourceRegister 中找到示例:

org.apache.spark.sql.execution.datasources.csv.CSVFileFormat
org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider
org.apache.spark.sql.execution.datasources.json.JsonFileFormat
org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
org.apache.spark.sql.execution.datasources.text.TextFileFormat
org.apache.spark.sql.execution.streaming.ConsoleSinkProvider
org.apache.spark.sql.execution.streaming.TextSocketSourceProvider
org.apache.spark.sql.execution.streaming.RateSourceProvider

这是链接 format 中的短名称的文件到实现。

我通常建议人们在 Spark 研讨会期间做的是从双方开始开发:

编写流查询(使用 format )，例如

val input = spark
  .readStream
  .format("yourCustomSource") // <-- your custom source here
  .load

实现流式传输 Source和相应的 DataSourceRegister (可能是同一个类(class))

(可选)注册 DataSourceRegister通过编写完全限定的类名，比如 com.mycompany.spark.MyDataSourceRegister , 至 META-INF/services/org.apache.spark.sql.sources.DataSourceRegister :

$ cat META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
com.mycompany.spark.MyDataSourceRegister

注册 DataSourceRegister 的最后一步自定义实现 Source是可选的，仅用于注册最终用户在 DataFrameReader.format 中使用的数据源别名。方法。

format(source: String): DataFrameReader Specifies the input data source format.

查看 org.apache.spark.sql.execution.streaming.RateSourceProvider的代码良好的开端。