devbox
代码探险家
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

spark streaming demo （java 1.8）

作者：代码探险家 | 2024-08-08 02:07:13

赞

踩

spark streaming demo

spark streaming demo 使用java 1.8 

从kafka获取数据， 写入redis的一个简单demo

 

发送给kafka的数据包：

{'type': 'Data', 'values': [{'compID': '3333', 'name': '冷却油压力', 'serial_num': '1', 'time': '2019-06-18 16:49:51', 'gateMac': 'D4:36:39:1A:0D:D3', 'function_code': '3', 'data': '7'}, {'compID': '3333', 'name': '工作油压力', 'serial_num': '1', 'time': '2019-06-18 16:49:51', 'gateMac': 'D4:36:39:1A:0D:D3', 'function_code': '3', 'data': '76'}]}

 

StreamingKafka.java
  1. package com.imooc.spark;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONArray;
  5. import com.alibaba.fastjson.JSONObject;
  6. import com.google.gson.Gson;
  7. import org.apache.kafka.clients.consumer.ConsumerRecord;
  8. import org.apache.kafka.common.TopicPartition;
  9. import org.apache.log4j.Logger;
  10. import org.apache.spark.SparkConf;
  11. import org.apache.spark.api.java.JavaRDD;
  12. import org.apache.spark.api.java.function.FlatMapFunction;
  13. import org.apache.spark.api.java.function.Function;
  14. import org.apache.spark.streaming.Durations;
  15. import org.apache.spark.streaming.StreamingContext;
  16. import org.apache.spark.streaming.api.java.*;
  17. import org.apache.spark.streaming.kafka010.ConsumerStrategies;
  18. import org.apache.spark.streaming.kafka010.KafkaUtils;
  19. import org.apache.spark.streaming.kafka010.LocationStrategies;
  20.  
  21. import redis.clients.jedis.Jedis;
  22. import scala.Tuple2;
  23.  
  24. import java.sql.Connection;
  25. import java.text.SimpleDateFormat;
  26. import java.util.*;
  27.  
  28. public class StreamingKafka {
  29.     public static void  main(String[] args) throws Exception{
  30.  
  31.         System.setProperty("hadoop.home.dir", "G:\\winutils\\");
  32.  
  33.         String topics = "yu1";
  34.         String brokers = "192.168.3.101:9093";
  35.         SparkConf conf = new SparkConf().setMaster("local[2]")
  36.                 .setAppName("StreamingKafka")
  37.                 .set("spark.rdd.compress", "true")
  38.                 .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
  39.         JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
  40.  
  41.         Collection<String> topicSet = new HashSet<>(Arrays.asList(topics.split(",")));
  42.         Map<String, Object> kafkaParams = new HashMap<>();
  43.         kafkaParams.put("bootstrap.servers", brokers);
  44.         kafkaParams.put("group.id", "group11111111");
  45.         kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  46.         kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  47.         kafkaParams.put("auto.offset.reset", "latest");
  48.  
  49. //        Map<TopicPartition, Long> offsets = new HashMap<>();
  50. //        offsets.put(new TopicPartition("yu_test_Data", 0), 2L);
  51.  
  52.         final JavaInputDStream<ConsumerRecord<String, String>> lines = KafkaUtils.createDirectStream(
  53.                 jssc,
  54.                 LocationStrategies.PreferConsistent(),
  55.                 ConsumerStrategies.Subscribe(topicSet, kafkaParams));
  56.  
  57.         /**
  58.          *  数据清洗
  59.          */
  60.         JavaDStream<deviceData> dstream = lines
  61.                 .flatMap(new FlatMapFunction<ConsumerRecord<String, String>, deviceData>() {
  62.                     @Override
  63.                     public Iterator<deviceData> call(ConsumerRecord<String, String> stringStringConsumerRecord) throws Exception {
  64.                         String message = stringStringConsumerRecord.value().replaceAll("\"", "");
  65.                         return DataHandle(message).iterator();
  66.                     }
  67.                 });
  68.  
  69.         /**
  70.          * 数据存储
  71.          */
  72.         dstream.foreachRDD(rdd -> {
  73.             rdd.foreach(record->{
  74.                 Jedis jedis = new Jedis("192.168.3.176");
  75.                 jedis.hset(record.compID + "/" + record.gateMac,  record.operationValue, record.operationData);
  76.             });
  77.         });
  78.  
  79.         dstream.print();
  80.  
  81.         jssc.start();
  82.         jssc.awaitTermination();
  83.     }
  84.  
  85.     /**
  86.      * 解析Data类型的数据
  87.      */
  88.     private static List<deviceData> DataHandle(String message) {
  89.         List<deviceData> d = new ArrayList<>();
  90.         try {
  91.             JSONObject DataObject = JSON.parseObject(message);
  92.             String dataType = (String) DataObject.get("type");
  93.             if (dataType.equals("Data") || dataType.equals("data")) {
  94.  
  95.                 JSONArray dataList = JSON.parseArray(DataObject.get("values").toString());
  96.                 for (int i = 0; i < dataList.size(); i++) {
  97.                     deviceData d1 = new deviceData();
  98.                     JSONObject dataDict = dataList.getJSONObject(i);
  99.                     d1.compID = dataDict.get("compID").toString();
  100.                     d1.gateMac = dataDict.get("gateMac").toString();
  101.                     d1.operationValue = dataDict.get("name").toString();
  102.                     d1.operationData = dataDict.get("data").toString();
  103.                     d1.gatherTime = dataDict.get("time").toString();
  104.                     d.add(d1);
  105.                 }
  106.             }
  107.         } catch (Throwable t) {
  108.             t.printStackTrace();
  109.         }
  110.         return d;
  111.     }
  112. }
deviceData.java 
  1. package com.imooc.spark;
  2.  
  3. /**
  4.  *   设备数据的数据结构
  5.  */
  6. class deviceData {
  7.     String compID;
  8.     String gateMac;
  9.     String operationValue;
  10.     String operationData;
  11.     String gatherTime;
  12.     long timestamp;
  13.  
  14.     @Override
  15.     public String toString() {
  16.         return "compID: " + compID + " operationValue: " + operationValue + " operationData: " + operationData;
  17.     }
  18. }

运行代码显示如下：

使用python脚本模拟发包

  1. from kafka import KafkaProducer
  2. import json
  3. import time
  4. import random
  5. import threading
  6.  
  7. '''
  8.     生产者demo
  9.     向test_lyl2主题中循环写入10条json数据
  10.     注意事项：要写入json数据需加上value_serializer参数，如下代码
  11. '''
  12.  
  13. sj = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  14.  
  15.  
  16. def send_msg(Num):
  17.     producer = KafkaProducer(
  18.                             value_serializer=lambda v: json.dumps(v).encode('utf-8'),
  19.                             bootstrap_servers=['192.168.3.101:9093']
  20.                          )
  21.     for i in range(Num):
  22.         time.sleep(4)
  23.         payload = dict()
  24.         payload["type"] = "Data"
  25.         payload["values"] = []
  26.         a = random.randint(5, 10)
  27.         payload["values"].append(
  28.             {"function_code": "3", "serial_num": "1", "data": "%s" % a,
  29.              "time": "%s" % sj, "name": "冷却油压力", "compID": "3333", "gateMac": "D4:36:39:1A:0D:D3"})
  30.         a = random.randint(40, 100)
  31.         # a = 0
  32.         payload["values"].append(
  33.             {"function_code": "3", "serial_num": "1", "data": "%s" % a,
  34.              "time": "%s" % sj, "name": "工作油压力", "compID": "3333", "gateMac": "D4:36:39:1A:0D:D3"})
  35.         a = random.randint(0, 20)
  36.         # a = 0
  37.         payload["values"].append(
  38.             {"function_code": "3", "serial_num": "1", "data": "%s" % a,
  39.              "time": "%s" % sj, "name": "冷却水温度", "compID": "3333", "gateMac": "D4:36:39:1A:0D:D3"})
  40.         a = random.randint(50, 60)
  41.         payload["values"].append(
  42.             {"function_code": "3", "serial_num": "1", "data": "%s" % a,
  43.              "time": "%s" % sj, "name": "液位", "compID": "3333", "gateMac": "D4:36:39:1A:0D:D3"})
  44.         a = random.randint(1, 20)
  45.         # a = 0
  46.         payload["values"].append(
  47.             {"data": "%s" % a, "time": "%s" % sj, "name": "转速", "compID": "3333", "gateMac": "D4:36:39:1A:0D:D3"})
  48.         print("==", str(payload))
  49.         producer.send('yu1', str(payload))
  50.         print("============%s" % i)
  51.     end = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
  52.     print("===开始时间：%s" % sj)
  53.     print("=====截止时间： %s" % end)
  54.     producer.close()
  55.  
  56.  
  57. # def thread_start():
  58. #     num = 4
  59. #     Num = 1000000
  60. #     Threads = []
  61. #
  62. #     for i in range(num):
  63. #         Threads.append(threading.Thread(target=send_msg, args=(Num,)))
  64. #     for t in Threads:
  65. #         # t.setDaemon(True)
  66. #         t.start()
  67.  
  68. if __name__ == "__main__":
  69.     # thread_start()
  70.     send_msg(10000000)
  71.

 

pom.xml

  1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  2.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  3.     <modelVersion>4.0.0</modelVersion>
  4.     <groupId>com.imooc.spark</groupId>
  5.     <artifactId>sparktrain</artifactId>
  6.     <version>1.0</version>
  7.     <inceptionYear>2008</inceptionYear>
  8.     <properties>
  9.         <scala.version>2.11.12</scala.version>
  10.         <!--<kafka.version>0.9.0.0</kafka.version>-->
  11.         <spark.version>2.1.0</spark.version>
  12.         <hadoop.version>2.6.0-cdh5.7.0</hadoop.version>
  13.         <hbase.version>1.2.0-cdh5.7.0</hbase.version>
  14.     </properties>
  15.     <!--添加cloudera的repository-->
  16.     <repositories>
  17.         <repository>
  18.             <id>cloudera</id>
  19.             <url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
  20.         </repository>
  21.     </repositories>
  22.  
  23.     <dependencies>
  24.         <dependency>
  25.             <groupId>org.scala-lang</groupId>
  26.             <artifactId>scala-library</artifactId>
  27.             <version>${scala.version}</version>
  28.         </dependency>
  29.  
  30.         <!-- Kafka 依赖-->
  31.         <!--
  32.         <dependency>
  33.             <groupId>org.apache.kafka</groupId>
  34.             <artifactId>kafka_2.11</artifactId>
  35.             <version>${kafka.version}</version>
  36.         </dependency>
  37.         -->
  38.  
  39.         <!-- Hadoop 依赖-->
  40.         <dependency>
  41.             <groupId>org.apache.hadoop</groupId>
  42.             <artifactId>hadoop-client</artifactId>
  43.             <version>${hadoop.version}</version>
  44.         </dependency>
  45.  
  46.         <!-- HBase 依赖-->
  47.         <dependency>
  48.             <groupId>org.apache.hbase</groupId>
  49.             <artifactId>hbase-client</artifactId>
  50.             <version>${hbase.version}</version>
  51.         </dependency>
  52.  
  53.         <dependency>
  54.             <groupId>org.apache.hbase</groupId>
  55.             <artifactId>hbase-server</artifactId>
  56.             <version>${hbase.version}</version>
  57.         </dependency>
  58.  
  59.         <!-- Spark Streaming 依赖-->
  60.         <dependency>
  61.             <groupId>org.apache.spark</groupId>
  62.             <artifactId>spark-streaming_2.11</artifactId>
  63.             <version>${spark.version}</version>
  64.         </dependency>
  65.  
  66.         <dependency>
  67.             <groupId>com.alibaba</groupId>
  68.             <artifactId>fastjson</artifactId>
  69.             <version>1.2.47</version>
  70.         </dependency>
  71.         <dependency>
  72.             <groupId>com.google.code.gson</groupId>
  73.             <artifactId>gson</artifactId>
  74.             <version>2.2.4</version>
  75.         </dependency>
  76.         <dependency>
  77.             <groupId>redis.clients</groupId>
  78.             <artifactId>jedis</artifactId>
  79.             <version>2.3.0</version>
  80.         </dependency>
  81.  
  82.  
  83.         <!-- Spark Streaming整合Flume 依赖-->
  84.         <dependency>
  85.             <groupId>org.apache.spark</groupId>
  86.             <artifactId>spark-streaming-flume_2.11</artifactId>
  87.             <version>${spark.version}</version>
  88.         </dependency>
  89.  
  90.         <dependency>
  91.             <groupId>org.apache.spark</groupId>
  92.             <artifactId>spark-streaming-flume-sink_2.11</artifactId>
  93.             <version>${spark.version}</version>
  94.         </dependency>
  95.  
  96.         <dependency>
  97.             <groupId>org.apache.spark</groupId>
  98.             <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
  99.             <version>${spark.version}</version>
  100.         </dependency>
  101.  
  102.         <dependency>
  103.             <groupId>org.apache.commons</groupId>
  104.             <artifactId>commons-lang3</artifactId>
  105.             <version>3.5</version>
  106.         </dependency>
  107.  
  108.         <!-- Spark SQL 依赖-->
  109.         <dependency>
  110.             <groupId>org.apache.spark</groupId>
  111.             <artifactId>spark-sql_2.11</artifactId>
  112.             <version>${spark.version}</version>
  113.         </dependency>
  114.  
  115.  
  116.         <dependency>
  117.             <groupId>com.fasterxml.jackson.module</groupId>
  118.             <artifactId>jackson-module-scala_2.11</artifactId>
  119.             <version>2.6.5</version>
  120.         </dependency>
  121.  
  122.         <dependency>
  123.             <groupId>net.jpountz.lz4</groupId>
  124.             <artifactId>lz4</artifactId>
  125.             <version>1.3.0</version>
  126.         </dependency>
  127.  
  128.         <dependency>
  129.             <groupId>mysql</groupId>
  130.             <artifactId>mysql-connector-java</artifactId>
  131.             <version>5.1.38</version>
  132.         </dependency>
  133.  
  134.         <dependency>
  135.             <groupId>org.apache.flume.flume-ng-clients</groupId>
  136.             <artifactId>flume-ng-log4jappender</artifactId>
  137.             <version>1.6.0</version>
  138.         </dependency>
  139.  
  140.     </dependencies>
  141.  
  142.     <build>
  143.         <!--
  144.         <sourceDirectory>src/main/scala</sourceDirectory>
  145.         <testSourceDirectory>src/test/scala</testSourceDirectory>
  146.         -->
  147.         <plugins>
  148.             <plugin>
  149.                 <groupId>org.scala-tools</groupId>
  150.                 <artifactId>maven-scala-plugin</artifactId>
  151.                 <executions>
  152.                     <execution>
  153.                         <goals>
  154.                             <goal>compile</goal>
  155.                             <goal>testCompile</goal>
  156.                         </goals>
  157.                     </execution>
  158.                 </executions>
  159.                 <configuration>
  160.                     <scalaVersion>${scala.version}</scalaVersion>
  161.                     <args>
  162.                         <arg>-target:jvm-1.8</arg>
  163.                     </args>
  164.                 </configuration>
  165.             </plugin>
  166.             <plugin>
  167.                 <groupId>org.apache.maven.plugins</groupId>
  168.                 <artifactId>maven-eclipse-plugin</artifactId>
  169.                 <configuration>
  170.                     <source>8</source>
  171.                     <target>8</target>
  172.                     <downloadSources>true</downloadSources>
  173.                     <buildcommands>
  174.                         <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>
  175.                     </buildcommands>
  176.                     <additionalProjectnatures>
  177.                         <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>
  178.                     </additionalProjectnatures>
  179.                     <classpathContainers>
  180.                         <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
  181.                         <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>
  182.                     </classpathContainers>
  183.                 </configuration>
  184.             </plugin>
  185.         </plugins>
  186.     </build>
  187.     <reporting>
  188.         <plugins>
  189.             <plugin>
  190.                 <groupId>org.scala-tools</groupId>
  191.                 <artifactId>maven-scala-plugin</artifactId>
  192.                 <configuration>
  193.                     <scalaVersion>${scala.version}</scalaVersion>
  194.                 </configuration>
  195.             </plugin>
  196.         </plugins>
  197.     </reporting>
  198. </project>

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/代码探险家/article/detail/945578
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号