没有输出到Kafka主题：Spark Structured Streaming和Kafka Integration

Question

我正在尝试使用kafka接收器将Apache Spark 2.3.1的流输出发送到Apache Kafka：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.udf
import org.apache.kafka.clients
import org.apache.spark.streaming
import java.sql.Timestamp
import java.util.Properties

object CQ3D {    
  def main(args: Array[String]) {

val spark = SparkSession
      .builder
      .appName("test")
      .getOrCreate()

val predictionStreamSchema = new StructType()
  .add("production_id", "long")
  .add("type", "string")

val lines = spark
      .readStream
      .option("sep", ",")
      .schema(testSchema)
      .csv("/path/to/directory/")

val query = lines.selectExpr("CAST(production_id AS STRING) AS key", "type AS value").writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("topic", "test")
      .option("checkpointLocation", "/local/directory")
      .outputMode("complete")
      .start()

query.awaitTermination()

我的build.sbt文件看起来像：

name := "CQ3D"
version := "0.1"
scalaVersion := "2.11.8"
val sparkVersion = "2.3.1"
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-sql" % sparkVersion,
"org.apache.spark" %% "spark-streaming" % sparkVersion,
"org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVersion
)

libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % sparkVersion

我的代码使用控制台接收器提供正确的输出，但是在使用kafka接收器时没有生成输出或发送到kafka主题。我的kafka zookeeper和kafka服务器在同一台机器上运行。控制台消息如下：

./spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 --class CQ3D --master local[4] /home/salman/Development

/SparkStreaming/Scala/target/scala-2.11/cq3d_2.11-0.1.jar
Ivy Default Cache set to: /home/salman/.ivy2/cache
The jars for the packages stored in: /home/salman/.ivy2/jars
:: loading settings :: url = jar:file:/home/salman/spark-2.3.1-bin-hadoop2.7/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
org.apache.spark#spark-sql-kafka-0-10_2.11 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-18e5a4df-cae8-4cf2-92bb-e02af7673888;1.0
    confs: [default]
    found org.apache.spark#spark-sql-kafka-0-10_2.11;2.3.1 in spark-list
    found org.apache.kafka#kafka-clients;0.10.0.1 in spark-list
    found net.jpountz.lz4#lz4;1.3.0 in spark-list
    found org.xerial.snappy#snappy-java;1.1.2.6 in spark-list
    found org.slf4j#slf4j-api;1.7.21 in central
    found org.spark-project.spark#unused;1.0.0 in spark-list
:: resolution report :: resolve 247ms :: artifacts dl 4ms
    :: modules in use:
    net.jpountz.lz4#lz4;1.3.0 from spark-list in [default]
    org.apache.kafka#kafka-clients;0.10.0.1 from spark-list in [default]
    org.apache.spark#spark-sql-kafka-0-10_2.11;2.3.1 from spark-list in [default]
    org.slf4j#slf4j-api;1.7.21 from central in [default]
    org.spark-project.spark#unused;1.0.0 from spark-list in [default]
    org.xerial.snappy#snappy-java;1.1.2.6 from spark-list in [default]
    ---------------------------------------------------------------------
    |                  |            modules            ||   artifacts   |
    |       conf       | number| search|dwnlded|evicted|| number|dwnlded|
    ---------------------------------------------------------------------
    |      default     |   6   |   0   |   0   |   0   ||   6   |   0   |
    ---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent-18e5a4df-cae8-4cf2-92bb-e02af7673888
    confs: [default]
    0 artifacts copied, 6 already retrieved (0kB/5ms)
2018-09-14 20:14:58 WARN  Utils:66 - Your hostname, salman-ubuntu-desktop resolves to a loopback address: 127.0.1.1; using 150.82.219.122 instead (on interface enp4s0)
2018-09-14 20:14:58 WARN  Utils:66 - Set SPARK_LOCAL_IP if you need to bind to another address
2018-09-14 20:14:59 WARN  NativeCodeLoader:62 - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2018-09-14 20:14:59 INFO  SparkContext:54 - Running Spark version 2.3.1
2018-09-14 20:14:59 INFO  SparkContext:54 - Submitted application: CQ3D
2018-09-14 20:14:59 INFO  SecurityManager:54 - Changing view acls to: salman
2018-09-14 20:14:59 INFO  SecurityManager:54 - Changing modify acls to: salman
2018-09-14 20:14:59 INFO  SecurityManager:54 - Changing view acls groups to: 
2018-09-14 20:14:59 INFO  SecurityManager:54 - Changing modify acls groups to: 
2018-09-14 20:14:59 INFO  SecurityManager:54 - SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(salman); groups with view permissions: Set(); users  with modify permissions: Set(salman); groups with modify permissions: Set()
2018-09-14 20:14:59 INFO  Utils:54 - Successfully started service 'sparkDriver' on port 36805.

我使用正确的import和/或libraryDependencies吗？

有时在编译时我收到以下警告：

[warn] There may be incompatibilities among your library dependencies.
[warn] Run 'evicted' to see detailed eviction warnings

但是，代码仍然使用“sbt包”编译。当我使用以下代码执行代码时，我在kafka主题中没有得到任何输出？

./spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 --class testClass --master local[4] /home/user/Dev/Scala/target/scala-2.11/testClass_2.11-0.1.jar

Answer 1

在Spark文档中，它提到对于来自本地文件系统的Spark Streaming，文件必须以原子方式移动到源文件夹中。可能有一个配置来读取现有文件，但我不记得了。

在评论中，我提到了Kafka Connect，它是一个用于向Kafka传输数据的内置框架，您只需要构建链接项目并运行Kafka Connect。

否则，如果你已经使用Hadoop，我建议其他人使用Flume，如果你有Elasticsearch将文件带入Kafka，我建议使用Filebeat / Fluentd。基本上，Spark这么简单的程序从本地文件系统读取的开销太大，并且不需要任何并行性来读取每个文件