Проблема с нулевыми значениями при потоковой передаче через Spark и Kafka - PullRequest
Новая
       8

Проблема с нулевыми значениями при потоковой передаче через Spark и Kafka

0 голосов
/ 08 января 2019

Я создал SparkConsumer, чтобы я мог отправить файл CSV в Kafka через Spark Structured Streaming. Я запускаю sparkConsumer, затем он ждет продюсера. Я запускаю Producer, и файл отправляется. Проблема в том, что я становлюсь «нулевыми» значениями в кадре данных, а не в контенте. Мой вывод выглядит так: 

-------------------------------------------
Batch: 1
-------------------------------------------
+---------+---------+-----------+--------+-----------------------+
|InvoiceNo|StockCode|Description|Quantity|timestamp              |
+---------+---------+-----------+--------+-----------------------+
|null     |null     |null       |null    |2019-01-08 15:46:29.156|
|null     |null     |null       |null    |2019-01-08 15:46:29.224|
|null     |null     |null       |null    |2019-01-08 15:46:29.224|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.225|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
|null     |null     |null       |null    |2019-01-08 15:46:29.241|
+---------+---------+-----------+--------+-----------------------+

Код для sparkConsumer: 

object sparkConsumer extends App {

  val rootLogger = Logger.getRootLogger()
  rootLogger.setLevel(Level.ERROR)

  val spark = SparkSession
    .builder()
    .appName("Spark-Kafka-Integration")
    .master("local")
    .getOrCreate()

  val schema = StructType(Array(
    StructField("InvoiceNo", StringType, nullable = true),
    StructField("StockCode", StringType, nullable = true),
    StructField("Description", StringType, nullable = true),
    StructField("Quantity", StringType, nullable = true)
  ))

  import spark.implicits._
  val df = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "test")
    .option("delimiter", ";")
    .option("header","true")
    .option("inferSchema","true")
    .load()

  val df1 = df.selectExpr("CAST(value as STRING)", "CAST(timestamp AS TIMESTAMP)").as[(String, Timestamp)]
    .select(from_json($"value", schema).as("data"), $"timestamp")
    .select("data.*", "timestamp")


  df1.writeStream
    .format("console")
    .option("truncate","false")
    .start()
    .awaitTermination()

}

Producer.scala: 

object Producer extends App {
  import java.util.Properties
  import org.apache.kafka.clients.producer._

  val  props = new Properties()
  props.put("bootstrap.servers", "localhost:9092")                                             
  props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")        
  props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")      

  val producer = new KafkaProducer[String, String](props)                                             
  val TOPIC="test"
  val fileName = "path/to/test.csv"
  val lines = Source.fromFile(fileName).getLines()

  for(i <- lines){
    val record = new ProducerRecord(TOPIC, "key", s"$i")                    
    producer.send(record)
  }
  val record = new ProducerRecord(TOPIC, "key", "the end "+new java.util.Date)
  producer.send(record)
  producer.close()

}

Может ли кто-нибудь помочь мне стать содержимым моего файла?

1 Ответ

0 голосов
/ 16 января 2019

Я думаю, что проблема связана с сериализацией и десериализацией. Ваш value, который вы пишете в теме, имеет формат csv, например:

111,someCode,someDescription,11

Ваш Потребитель Spark считает, что сообщение в формате json (from_json с некоторой схемой). Если сообщение будет примерно таким, как показано ниже, синтаксический анализ будет работать. 

{
    "InvoiceNo": "111",
    "StockCode": "someCode",
    "Description": "someDescription",
    "Quantity": "11"
}

Вы должны изменить сериализацию или десериализацию, чтобы они соответствовали друг другу.

Один из следующих вариантов должен работать

  1. Производитель должен написать сообщение к теме в формате json
  2. Потребитель Spark должен проанализировать строку, используя comma для разделения полей
...