Публикация результата Apache Spark в другом приложении/Кафке

В настоящее время я разрабатываю модуль быстрой агрегации данных, который получает события и публикует их в кластере Kafka. Затем у нас есть интеграция Kafka и Spark Streaming. Spark Streaming считывает поток из Kafka и выполняет некоторые вычисления. Когда вычисление завершено, нам нужно отправить результат в другое приложение. Это приложение может быть веб-службой или кластером Kafka.

Мне интересно, как мы можем это сделать? Из того, что я читал, Spark Stream отправляет данные в нисходящие потоки, такие как базы данных и файловые системы.

Как бы вы поступили с дизайном такого приложения? Должен ли я заменить Spark Stream на Storm, чтобы иметь возможность публиковать результаты в другом приложении?


apache-kafka apache-spark spark-streaming apache-storm
person sansari    schedule 23.03.2016    source источник

Ответы (2)


arrow_upward
2
arrow_downward

Обратитесь к dstream.foreachRDD, мощному примитиву, который позволяет отправлять данные во внешние системы.
Шаблоны проектирования для использования foreachRDD

Ниже приведен мой код интеграции kafka для справки (не оптимизирован, только для POC, объект KafkaProducer можно повторно использовать в foreachRDD):

DStream.foreachRDD(rdd => {
      rdd.foreachPartition { partitionOfRecords =>
        val kafkaProps = new Properties()
        kafkaProps.put("bootstrap.servers", props("bootstrap.servers"))
        kafkaProps.put("client.id", "KafkaIntegration Producer");
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        val producer = new KafkaProducer[String, String](kafkaProps);

        partitionOfRecords.foreach(record => {
        val message = new ProducerRecord[String, String]("hdfs_log_test", record.asInstanceOf[String])
          producer.send(message)
        })
        producer.close()
      }
    })
person Shawn Guo    schedule 23.03.2016

arrow_upward
1
arrow_downward

Мне интересно, как мы можем это сделать? Из того, что я читал, Spark Stream отправляет данные в нисходящие потоки, такие как базы данных и файловые системы.

Spark не ограничивается HDFS или базами данных, вы можете инициализировать подключение к любому доступному внешнему ресурсу. Это может быть возврат к Kafka, RabbitMQ или WebService.

Если вы выполняете простое преобразование, такое как map, filter, reduceByKey и т. д., то использование DStream.foreachRDD подойдет. Если вы будете выполнять вычисления с отслеживанием состояния, такие как DStream.mapWithState, то, как только вы закончите обработку состояния, вы можете просто отправить данные в любую внешнюю службу.

Например, мы используем Kafka в качестве входного потока данных, а RabbitMQ и выходной после выполнения некоторых вычислений с отслеживанием состояния.

person Yuval Itzchakov    schedule 23.03.2016