sparkStreaming+kafka+hbase实战练习一

ゝ一世哀愁。 2022-06-09 23:42 159阅读 0赞

这个需求是是按照实际工作中的项目改写的一个例子。

业务需求： 1. 实时统计某市银行流水交易总额

2. 实时统计某市银行某地区的流水交易总额

3. 每隔10s统一次每个地区近一年内的交易总额

系统需求： 保证数据不丢失、系统可容错

分析：这是一个实时流处理系统的常见应用，在网页点击分析中也可以看见类似的例子，kafka具有高可扩展性、容错性、高吞吐量、高并发、分布式等特性，常用于实时流中做消息传输，同时也起到了消息缓冲的作用，因此在这个例子中使用kafka对接外部系统做消息的传输。关于流式处理系统现在常用的有sparkStreaming、Storm，Storm是一个真正的流式处理系统，每次处理处理一条消息，SparkStreaming 看做一个伪流式处理系统，每次处理一个很小batch的数据，它基于SparkCore ，因此继承了sparkCore的很多特性，例如：任务执行优化、内存分布式计算、可容错、迭代式计算等，除了可类似于sparkCore RDD的算子外，还提供了状态函数与窗口函数，可对接kafka 、flume、文件、监听端口等，在我们这个例子中使用SparkStreaming 正好能满足我们的要求。Hbase是一个分布式的KV存储系统，能够快速读写数据，在实时流中Hbase用于存储源报文再好不过了。

设计： 使用kafka对接外部系统，kafka客户端消息的传输有两种方式：sync与async，即同步方式与异步方式，同步方式表示没发送一条消息就提交一次，这种做法能够保证数据不丢失，异步方式是批量发送提交，当客户端的消息达到一定大小就将其提交给broker，这种做法性能高，为了保证数据的不丢失，采用同步方式提交，另外要求该topic的所有副本都收到消息才算成功；kafka与SparkStreaming对接提供了两种方式，一种是reciver，一种是direct，这两种方式都可以做到数据容错，但是reciver方式代价更大，需要预写日志、在内存序列化中保证两份(默认)，而direct方式直接读取kafka中消息，使用消息offset手动提交，只有在处理成功的情况下才去读取下一条消息，因此在这里使用direct方式获取消息。我们要求系统能够7\*24小时不间断的工作，那就是保证系统的容错性，即时系统出现故障，也能够即时恢复到最近的状态，sparkStreaming中提供了checkpoint 机制，可保存系统的元数据信息与数据信息，恢复时直接读取元数据信息与数据信息恢复到最近的状态。统计流水交易总额与某地区的流水交易总额使用updateStateByKey，每隔10s统一次每个地区近一年内的交易总额使用reduceByKeyAndWindow算子操作。对于将源始报文保存到hbase，直接使用hbase api。

实现：

启动服务：hdfs、yarn、hbase、kafka

创建topic :  ./kafka-topics.sh  --zookeeper hadoop1:2181 --partitions 1 --replication-factor 1 --create --topic  topicName(只有一台服务器创建一个分区一个副本)

涉及四个topic 1. 源报文:topic2  2. 交易总额：topic\_all

创建hbase表：create 'trade','cf'

3 . 地区交易总额 ：topic\_regionsum 4. 每隔10s每个地区近一年内的交易总额 topic\_regionsumOneYear

创建maven java project项目，使用maven添加依赖：

<dependencies>  
    <dependency>  
      <groupId>junit</groupId>  
      <artifactId>junit</artifactId>  
      <version>3.8.1</version>  
      <scope>test</scope>  
    </dependency>  
    <dependency>  
    <groupId>org.apache.spark</groupId>  
    <artifactId>spark-streaming\_2.11</artifactId>  
    <version>2.2.0</version>  
</dependency>  
<dependency>  
    <groupId>org.apache.spark</groupId>  
    <artifactId>spark-sql\_2.10</artifactId>  
    <version>2.2.0</version>  
</dependency>  
    <dependency>  
       <groupId>org.apache.spark</groupId>  
       <artifactId>spark-streaming-kafka-0-8\_2.11</artifactId>  
       <version>2.2.0</version>  
    </dependency>  
    <dependency>  
   <groupId>org.apache.kafka</groupId>  
   <artifactId>kafka-clients</artifactId>  
   <version>0.8.2.0</version>  
</dependency>  
   <dependency>  
    <groupId>org.apache.hbase</groupId>  
    <artifactId>hbase-client</artifactId>  
    <version>1.3.1</version>  
   </dependency>  
   <dependency>  
        <groupId>org.apache.hbase</groupId>  
        <artifactId>hbase-server</artifactId>  
        <version>1.3.1</version>  
    </dependency>  
    <dependency>    
    <groupId>jdk.tools</groupId>    
    <artifactId>jdk.tools</artifactId>    
    <version>1.6</version>  
    <scope>system</scope>    
     <systemPath>$\{JAVA\_HOME\}/lib/tools.jar</systemPath>    
    </dependency>  
<dependency>  
    <groupId>com.google.code.gson</groupId>  
    <artifactId>gson</artifactId>  
    <version>2.2.4</version>  
</dependency>  
  </dependencies>

编码：

public class Demo2 \{  
private static String SPARK\_MASTER="yarn";    //使用yarn作为资源框架  
private static String SPARK\_DEPLOY\_MODE="client";  
private static String SPARK\_APP\_NAME="sparkStreaming";  
private static long SPARK\_BATCH\_SIZE\_M=10000;  //批处理间隔10s  
private static long SPARK\_WIN\_LEN=120000;   //窗口大小 2min （方便演示）  
private static long SPARK\_WIN\_SLID=10000;    //窗口滑动间隔10s

//checkpoint的路径  
private static String SPARK\_CHECKPOINT\_DIR="hdfs://hadoop1:9000/spark/streaming/checkpoint3";  
  
private static String KAFKA\_BOOTSTRAP\_SERVERS="hadoop1:9092";  
private static String KAFKA\_KEY\_DESERIALIZER="org.apache.kafka.common.serialization.StringDeserializer";  
private static String KAFKA\_VALUE\_DESERIALIZER="org.apache.kafka.common.serialization.StringDeserializer";  
private static String KAFKA\_GROUP\_ID="g1";  
private static String KAFKA\_AUTO\_OFFSET\_RESET="largest";  
private static String KAFKA\_ENABLE\_AUTO\_COMMIT="false";  
private static String KAFKA\_TOPIC="topic2";  
  
  
private static Function0<JavaStreamingContext> createStreamingContext=new Function0<JavaStreamingContext>() \{  
  
public JavaStreamingContext call() throws Exception \{  
  
SparkConf conf=new SparkConf();  
conf.set("spark.app.name", SPARK\_APP\_NAME);  
       conf.set("spark.master", SPARK\_MASTER);  
       conf.set("spark.submit.deployMode", SPARK\_DEPLOY\_MODE);  
       final JavaSparkContext jsc=new JavaSparkContext(conf);  
       JavaStreamingContext jssc=new JavaStreamingContext(jsc,Duration.apply(SPARK\_BATCH\_SIZE\_M));  
  
       //checkpoint  
       jssc.checkpoint(SPARK\_CHECKPOINT\_DIR);  
         
       //初始区域交易金额总和  
       List<Tuple2<String,Long>> initRegionSumData=  
        Arrays.asList(new Tuple2<String,Long>("A01", Long.valueOf(500000)),new Tuple2<String,Long>("A02", Long.valueOf(8000000)),  
        new Tuple2<String,Long>("A03", Long.valueOf(15000000)),new Tuple2<String,Long>("A04", Long.valueOf(6000000)));  
         
       //一定要为JavaPairRDD类型  
       JavaPairRDD<String,Long> initRegionSumRDD=jssc.sparkContext().parallelizePairs(initRegionSumData);

//初始交易总额  
       final long initAllSum=100000000;  
        
       List<Tuple2<String,Long>> initAllSumData=Arrays.asList(new Tuple2<String,Long>("all",initAllSum));  
         
       JavaPairRDD<String,Long> initAllSumRDD =jssc.sparkContext().parallelizePairs(initAllSumData);  
         
       Map<String,String> kafkaParams=new HashMap<String,String>();  
       kafkaParams.put("bootstrap.servers", KAFKA\_BOOTSTRAP\_SERVERS);  
       kafkaParams.put("key.deserializer", KAFKA\_KEY\_DESERIALIZER);  
       kafkaParams.put("value.deserializer", KAFKA\_VALUE\_DESERIALIZER);  
       kafkaParams.put("group.id", KAFKA\_GROUP\_ID);  
       kafkaParams.put("auto.offset.reset", KAFKA\_AUTO\_OFFSET\_RESET);  
       kafkaParams.put("enable.auto.commit", KAFKA\_ENABLE\_AUTO\_COMMIT);  
  
  
       Set<String> topics=new HashSet<String>();  
       topics.add(KAFKA\_TOPIC);  
         
      JavaPairInputDStream<String, String> kafkaS= KafkaUtils.createDirectStream(jssc, String.class,  
        String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics);  
         
      //过虑key  从kafka中获取的消息 key对于我们说是没有用的  
     JavaDStream<String> jds= kafkaS.map(new Function<Tuple2<String,String>, String>() \{  
public String call(Tuple2<String, String> v1) throws Exception \{  
return v1.\_2();  
\}  
 \});  
        
     //原始数据存入hbase中，源始报文使用tab制表符分割字段

//格式  数据格式：交易流水号交易地区编码交易类型交易金额交易时间

//例如： 00001A01T0120002017-08-01  
     jds.foreachRDD(new VoidFunction<JavaRDD<String>>() \{  
  
public void call(JavaRDD<String> t) throws Exception \{

//按照官网这里应该使用数据连接池，对每一个partition使用一个连接即可，后续优化  
t.foreachPartition(new VoidFunction<Iterator<String>>() \{  
  
public void call(Iterator<String> t) throws Exception \{  
  
  
  List<Put> listPut=new ArrayList<Put>();  
   while(t.hasNext())  
   \{  
String one=t.next();  
byte\[\] f=Bytes.toBytes("cf");  
String s\[\]=one.split("");  
ImmutableBytesWritable key=new ImmutableBytesWritable(Bytes.toBytes(s\[0\]));  
Put p=new Put(Bytes.toBytes(s\[0\]));  
p.addColumn(f, Bytes.toBytes("tradeId"), Bytes.toBytes(s\[0\]));  
p.addColumn(f, Bytes.toBytes("tradeAreaId"), Bytes.toBytes(s\[1\]));  
p.addColumn(f, Bytes.toBytes("tradeType"), Bytes.toBytes(s\[2\]));  
p.addColumn(f, Bytes.toBytes("tradeMoney"), Bytes.toBytes(s\[3\]));  
p.addColumn(f, Bytes.toBytes("tradeTime"), Bytes.toBytes(s\[4\]));  
listPut.add(p);  
   \}  
     
   Configuration conf=HBaseConfiguration.create();  
   conf.set("hbase.zookeeper.quorum", "hadoop1:2181");  
   HTable table=new HTable(conf, "trade");  
   table.put(listPut);  
                       table.flushCommits();  
                       table.close();  
\}  
\});  
\}  
\});  
      
     //返回<地区，交易金额>  
     JavaPairDStream<String, Long> jpds= jds.mapToPair(new PairFunction<String, String, Long>() \{  
  
  
public Tuple2<String, Long> call(String t) throws Exception \{  
  
  
String\[\] s=t.split("");  
  
return new Tuple2(s\[1\],Long.valueOf(s\[3\]));  
\}  
\});  
       
     //多次使用  缓存起来  
     jpds.cache();  
        
     //装换为key相同的交易记录（"all",金额）  
    JavaPairDStream<String, Long> jpdsMap= jpds.mapToPair(new PairFunction<Tuple2<String,Long>, String, Long>() \{  
  
public Tuple2<String, Long> call(Tuple2<String, Long> t) throws Exception \{  
// TODO Auto-generated method stub  
return new Tuple2("all",t.\_2());  
\}  
\});  
       
    //计算总的交易金额，使用默认分区与初始值  
   JavaPairDStream<String,Long> allResult= jpdsMap.updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() \{  
  
                       //v1表示新进来的数据 v2表示当前的数据  
public Optional<Long> call(List<Long> v1, Optional<Long> v2) throws Exception \{  
                   
  
Long news=0L;  
if(v2.isPresent())  
\{  
news=v2.get();  
\}  
for(Long i : v1)  
\{  
news+=i;  
\}  
  
return Optional.of(news);  
\}  
\},new HashPartitioner(jssc.sparkContext().defaultParallelism()), initAllSumRDD);  
      
     //打印  
   System.out.println("======总交易金额====");  
   allResult.print();  
     
   //发送到kafka中  
  sendMessageToKafka("topic\_all",allResult);     
     
   //统计地区交易总和 使用initRegionSumRDD做为初始值  
  JavaPairDStream<String, Long> regionSum=jpds.updateStateByKey(new Function2<List<Long>, Optional<Long>, Optional<Long>>() \{  
  
  
public Optional<Long> call(List<Long> v1, Optional<Long> v2) throws Exception \{  
Long news=0L;  
if(v2.isPresent())  
\{  
news=v2.get();  
\}  
for(Long i : v1)  
\{  
news+=i;  
\}  
return Optional.of(news);  
\}  
\}, new HashPartitioner(jssc.sparkContext().defaultParallelism()), initRegionSumRDD);  
     
  //打印  
  System.out.println("======地区交易总金额====");  
  regionSum.print(1);  
 sendMessageToKafka("topic\_regionsum",regionSum);  
    
    
   //统计各个地区两年内的交易总额  
JavaPairDStream<String, Long> regionSumInTwoYears= jpds.reduceByKeyAndWindow(new Function2<Long, Long, Long>() \{  
  
  
public Long call(Long v1, Long v2) throws Exception \{  
// TODO Auto-generated method stub  
return v1+v2;  
\}  
\}, Duration.apply(SPARK\_WIN\_LEN), Duration.apply(SPARK\_WIN\_SLID));  
    
System.out.println("======地区交易两年内总金额====");  
regionSumInTwoYears.print();       
  
  
   
   sendMessageToKafka("topic\_regionsumTwoYear",regionSumInTwoYears);  
   
return jssc;  
\}  
\};  
  
  
//将消息发送到kafka中   以json格式发送  
private static void sendMessageToKafka(final String topic,JavaPairDStream<String, Long> ds)  
\{  
ds.foreachRDD(new VoidFunction<JavaPairRDD<String,Long>>() \{  
  
  
public void call(JavaPairRDD<String, Long> t) throws Exception \{  
List<Tuple2<String, Long>> list= t.collect();  
  
  
Map<String,String> message=new HashMap<String,String>();  
                for(Tuple2<String, Long> tp: list)  
                \{  
                message.put(tp.\_1(), String.valueOf(tp.\_2()));  
                \}  
                  
                KafKaProducerUtil.sendMassage(topic,null,new Gson().toJson(message));  
\}  
\});  
\}  
  
  
public static void main(String\[\] args) \{  
        //这个做checkpoint 重启仍然能够回到最新一次的状态，还能保证数据不丢失  
 JavaStreamingContextjssc=JavaStreamingContext.getOrCreate(SPARK\_CHECKPOINT\_DIR,            createStreamingContext);  
 jssc.start();  
 try \{  
jssc.awaitTermination();  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
  
\}  
  
\}

public class KafKaProducerUtil \{  
  
  
private static String KAFKA\_METADATA\_BROKER\_LIST="hadoop1:9092";  
private static String KAFKA\_KEY\_SERIALIZER\_CLASS="kafka.serializer.StringEncoder";  
private static String KAFKA\_SERIALIZER\_CLASS="kafka.serializer.StringEncoder";  
private static String KAFKA\_REQUEST\_REQUIRED\_ACKS="1";  
private static String KAFKA\_COMPRESSION\_CODEC="snappy";  
private static String KAFKA\_MESSAGE\_SEND\_MAX\_RETRIES="2";  
private static String KAFKA\_RECONNECT\_BACKOFF\_MS="100";  
private static String KAFKA\_REQUEST\_TIMEOUT\_MS="10000";  
private static String KAFKA\_PRODUCER\_TYPE="sync"; //同步处理  
private static String KAFKA\_QUEUE\_BUFFERING\_MAX\_MS=""; //消息缓存最长时间 异步方式下配置  
private static String KAFKA\_QUEUE\_BUFFERING\_MAX\_MESSAGES="";//缓存消息的最大值  
private static String KAFKA\_QUEUE\_ENQUEUE\_TIMEOUT\_MS="";//消息进入队列的的等待时间  
private static String KAFKA\_BATCH\_NUM\_MESSAGES="";//批量消息条数  
private static String KAFKA\_SEND\_BUFFER\_BYTES="";//socket缓存大小  
  
private static Producer<String, String> producer;  
  
private static Producer<String, String> getProducer()  
\{  
  
if(producer==null)  
\{  
synchronized (KafKaProducerUtil.class) \{  
Properties pro=new Properties();  
pro.setProperty("metadata.broker.list",KAFKA\_METADATA\_BROKER\_LIST);  
pro.setProperty("request.required.acks", KAFKA\_REQUEST\_REQUIRED\_ACKS);  
pro.setProperty("producer.type", KAFKA\_PRODUCER\_TYPE);  
pro.setProperty("serializer.class", KAFKA\_SERIALIZER\_CLASS);  
pro.setProperty("request.timeout.ms", KAFKA\_REQUEST\_TIMEOUT\_MS);  
pro.setProperty("key.serializer.class", KAFKA\_KEY\_SERIALIZER\_CLASS);  
pro.setProperty("compression.codec", KAFKA\_COMPRESSION\_CODEC);  
pro.setProperty("message.send.max.retries", KAFKA\_MESSAGE\_SEND\_MAX\_RETRIES);  
pro.setProperty("retry.backoff.ms", KAFKA\_RECONNECT\_BACKOFF\_MS);  
ProducerConfig config=new ProducerConfig(pro);  
producer=new  Producer<String, String>(config);  
\}  
  
\}  
  
return producer;  
\}  
  
  
public static void sendMassage(String topic,String key,String message)  
\{  
Producer<String, String> producer=getProducer();  
KeyedMessage<String, String> km=new KeyedMessage<String, String>(topic, key, message);  
producer.send(km);  
\}  
  
  
public static void sendListMassage(String topic,Map<String,String> msg)  
\{  
Producer<String, String> producer=getProducer();  
List<KeyedMessage<String, String>> list=new ArrayList<KeyedMessage<String, String>>();  
for(String k : msg.keySet())  
\{  
 KeyedMessage<String, String> km=new KeyedMessage<String, String>(topic, k, msg.get(k));  
 list.add(km);  
\}  
producer.send(list);  
\}

编码完成之后打成jar包，上传到服务器。需要将hbase的相关jar包放到spark的jars目录下，另外需要kafka与sparkStreaming的集成jar包spark-streaming-kafka-0-8-assembly\_2.11-2.0.2.jar，

启动服务： sh spark-submit  --class sparkStreaming.sparkStreaming.Demo2 /bigdata/spark/javademo/sparkStreamingDemo3.jar

生产者： ./kafka-console-producer.sh --broker-list hadoop1:9092 --topic topic2  （这里先用console）

消费者：./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_all

./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_regionsum

./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_regionsumTwoYear

结果：

![Image 1][]![Image 1][]     \[root@hadoop1 bin\]\# ./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_all  
    \{"all":"100000000"\}

\[root@hadoop1 bin\]\#  ./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_regionsum  
  \{"A02":"8000000","A01":"500000","A04":"6000000","A03":"15000000"\}

\[root@hadoop1 bin\]\#  ./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_regionsumTwoYear  
 \{\}

当我们生产消息时：

![Image 1][]![Image 1][]  \[root@hadoop1 bin\]\# ./kafka-console-producer.sh --broker-list hadoop1:9092 --topic topic2  
\[2017-08-05 01:53:17,575\] WARN Property topic is not valid (kafka.utils.VerifiableProperties)  
00001   A01     T01     2000    2017-08-01 12:00:00

\[root@hadoop1 bin\]\# ./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_all  
\{"all":"100000000"\}  
\{"all":"100000000"\}  
\{"all":"100002000"\}  
\{"all":"100002000"\}

\[root@hadoop1 bin\]\#  ./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_regionsum  
\{"A02":"8000000","A01":"500000","A04":"6000000","A03":"15000000"\}  
\{"A02":"8000000","A01":"500000","A04":"6000000","A03":"15000000"\}  
\{"A02":"8000000","A01":"502000","A04":"6000000","A03":"15000000"\}  
\{"A02":"8000000","A01":"502000","A04":"6000000","A03":"15000000"\}

\[root@hadoop1 bin\]\#  ./kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic topic\_regionsumTwoYear  
\{\}  
\{\}  
\{"A01":"2000"\}  
\{"A01":"2000"\}

hbase 中记录：

hbase(main):001:0> scan 'trade'  
ROW                                              COLUMN+CELL                                                                                                                                   
 00001                                           column=cf:tradeAreaId, timestamp=1501923210595, value=A01                                                                                     
 00001                                           column=cf:tradeId, timestamp=1501923210595, value=00001                                                                                       
 00001                                           column=cf:tradeMoney, timestamp=1501923210595, value=2000                                                                                     
 00001                                           column=cf:tradeTime, timestamp=1501923210595, value=2017-08-01 12:00:00                                                                       
 00001                                           column=cf:tradeType, timestamp=1501923210595, value=T01

![Image 1][]![Image 1][]

到这里整个实例主要流程已经演示完毕，后续将继续完成kafka生产者部分与消费者部分，研究在sparkStreaming中直接使用SQL方式将数据插入到hbase，研究如果在对原始报文需要关联查询，使用hbase还是使用spark的广播变量。

对于以上内容欢迎大脚拍砖~~~~~~~~~~~~~

[Image 1]: