sparkstreaming连接消费nsq

目的

• 使用 NSQ作为消息流
• 使用 spark-streaming 进行消费
• 对数据进行清洗后&＃xff0c;保存到hive仓库中

连接方案

1、编写Spark Streaming Custom Receivers&＃xff08;spark-streaming 自定义接收器&＃xff09;&＃xff0c;详细见文档

2、使用 nsq 官方提供的Java程序连接包 JavaNSQClient &＃xff0c;详细见文档

详细代码

自定义连接器

ReliableNSQReceiver.scala

import com.github.brainlag.nsq.callbacks.NSQMessageCallback
import com.github.brainlag.nsq.lookup.DefaultNSQLookup
import com.github.brainlag.nsq.{NSQConsumer, NSQMessage}
import org.apache.spark.internal.Logging
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiverclass MessageCallbacks(store_fun:String &＃61;> Unit) extends NSQMessageCallback with Logging {def message(message: NSQMessage): Unit &＃61;{val s &＃61; new String(message.getMessage())store_fun(s)message.finished()}
}
/* 自定义连接器 */
class ReliableNSQReceiver(host: String, port: Int, topic: String, channel: String)extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) with Logging {var consumer: NSQConsumer &＃61; nulldef onStart() {// 启动通过连接接收数据的线程new Thread("Socket Receiver") {override def run() { receive() }}.start()}def onStop() {logInfo("Stopped receiving")consumer.close}/** 接收数据 */private def receive() {try {val lookup &＃61; new DefaultNSQLookuplookup.addLookupAddress(host, port)consumer &＃61; new NSQConsumer(lookup, topic, channel, new MessageCallbacks(store))consumer.start} catch {case e: java.net.ConnectException &＃61;>restart("Error connecting to " &＃43; host &＃43; ":" &＃43; port, e)case t: Throwable &＃61;>restart("Error receiving data", t)}}}


使用连接器

import com.google.gson.JsonParser
import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}/*
* 在定义一个 context 之后,您必须执行以下操作.* 通过创建输入 DStreams 来定义输入源.
* 通过应用转换和输出操作 DStreams 定义流计算&＃xff08;streaming computations&＃xff09;.
* 开始接收输入并且使用 streamingContext.start() 来处理数据.
* 使用 streamingContext.awaitTermination() 等待处理被终止&＃xff08;手动或者由于任何错误&＃xff09;.
* 使用 streamingContext.stop() 来手动的停止处理.*/object ELKStreaming extends Logging{def main(args: Array[String]): Unit &＃61;{if (args.length <4) {System.err.println("Usage: ELKStreaming ")System.exit(1)}logInfo("start &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;>")StreamingExamples.setStreamingLogLevels()val sparkConf &＃61; new SparkConf().setAppName("ELKStreaming").setMaster("yarn").set("hive.metastore.uris", "thrift://hadoop15.bigdata.org:9083")// 创建一个批次间隔为10val ssc &＃61; new StreamingContext(sparkConf, Seconds(args(2).toInt))// 使用自定义的NSQReceiverval lines &＃61; ssc.receiverStream(new ReliableNSQReceiver(args(0), args(1).toInt, "log", "scalatest"))val hiveStream: DStream[(String, String)] &＃61; lines.map(line &＃61;> prefix_exit(line))// 将计算后的数据保存到hive中hiveStream.foreachRDD(rdd &＃61;> {// 利用SparkConf来初始化SparkSession。val sparkSession: SparkSession &＃61; SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()// 导入隐式转换来将RDDimport sparkSession.implicits._// 将RDD转换成DFval df: DataFrame &＃61; rdd.toDF("str", "ymd")// 取出表中的字段logInfo("df count &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;>"&＃43; df.count)df.createOrReplaceTempView("spark_logs")sparkSession.sql("insert into "&＃43;args(3)&＃43;" partition (ymd) select str,ymd from spark_logs")})ssc.start()ssc.awaitTermination()}def prefix_exit(line:String):(String,String) &＃61;{// 对数据进行清洗计算val obj &＃61; new JsonParser().parse(line).getAsJsonObjectval data_str1 &＃61; obj.get("recv_timestamp").toString().split("T|Z|\"")val data_str2 &＃61; data_str1(1).split(&＃39;-&＃39;)val data_str3 &＃61; data_str2(1)&＃43;"/"&＃43;data_str2(2)&＃43;"/"&＃43;data_str2(0)&＃43;" "&＃43;data_str1(2)&＃43;" [I] "&＃43;obj.get("index_type").toString().split("\"")(1)&＃43;" "&＃43;lineval data_str4 &＃61; data_str2(0)&＃43;data_str2(1)&＃43;data_str2(2)(data_str3.toString(), data_str4.toString())}
}