文档章节

Maven+Eclipse+SparkStreaming+Kafka整合

四叶草666
 四叶草666
发布于 2017/07/18 15:36
字数 1102
阅读 24
收藏 0

版本号:

maven3.5.0     scala IDE for Eclipse:版本(4.6.1)    spark-2.1.1-bin-hadoop2.7    kafka_2.11-0.8.2.1   JDK1.8

基础环境:

Maven3.5.0安装与配置+Eclipse应用

Maven下载项目依赖jar包和使用方法

maven中把依赖的JAR包一起打包

MAVEN Scope使用

一、指定JDK为1.8

在pom.xml配置文件中添加以下参数即可:

 
  1. <properties>
  2. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3. <encoding>UTF-8</encoding>
  4.     <java.version>1.8</java.version>
  5.     <maven.compiler.source>1.8</maven.compiler.source>
  6.     <maven.compiler.target>1.8</maven.compiler.target>
  7. </properties>
 
  1. <plugin>  
  2.     <groupId>org.apache.maven.plugins</groupId>  
  3.     <artifactId>maven-compiler-plugin</artifactId>  
  4.     <configuration>  
  5.         <source>1.8</source>  
  6.         <target>1.8</target>  
  7.     </configuration>  
  8. </plugin>

配置之后的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/xsd/maven-4.0.0.xsd">
  3. <modelVersion>4.0.0</modelVersion>
  4.  
  5. <groupId>Test</groupId>
  6. <artifactId>test</artifactId>
  7. <version>0.0.1-SNAPSHOT</version>
  8. <packaging>jar</packaging>
  9.  
  10. <name>test</name>
  11. <url>http://maven.apache.org</url>
  12.  
  13. <properties>
  14. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  15. <encoding>UTF-8</encoding>
  16. <!-- 配置JDK为1.8 -->
  17.         <java.version>1.8</java.version>
  18.         <maven.compiler.source>1.8</maven.compiler.source>
  19.         <maven.compiler.target>1.8</maven.compiler.target>
  20. </properties>
  21.  
  22. <dependencies>
  23. <dependency>
  24. <groupId>junit</groupId>
  25. <artifactId>junit</artifactId>
  26. <version>3.8.1</version>
  27. <scope>test</scope>
  28. </dependency>
  29. </dependencies>
  30.  
  31. <build>
  32. <plugins>
  33. <!-- 配置JDK为1.8 -->
  34. <plugin>  
  35.                 <groupId>org.apache.maven.plugins</groupId>  
  36.                 <artifactId>maven-compiler-plugin</artifactId>  
  37.                 <configuration>  
  38.                     <source>1.8</source>  
  39.                     <target>1.8</target>  
  40.                 </configuration>  
  41.             </plugin>  
  42.         
  43.              <!-- 配置打包依赖包maven-assembly-plugin -->
  44. <plugin>
  45. <artifactId> maven-assembly-plugin </artifactId>
  46. <configuration>
  47. <descriptorRefs>
  48. <descriptorRef>jar-with-dependencies</descriptorRef>
  49. </descriptorRefs>
  50. <archive>
  51. <manifest>
  52. <mainClass></mainClass>
  53. </manifest>
  54. </archive>
  55. </configuration>
  56. <executions>
  57. <execution>
  58. <id>make-assembly</id>
  59. <phase>package</phase>
  60. <goals>
  61. <goal>assembly</goal>
  62. </goals>
  63. </execution>
  64. </executions>
  65. </plugin>
  66. </plugins>
  67. </build>
  68. </project>

二、配置Spark依赖包

查看spark-2.1.1-bin-hadoop2.7/jars目录下的jar包版本

到maven远程仓库http://mvnrepository.com中搜索对应jar包即可。

1、配置spark-core_2.11-2.1.1.jar

往pom.xml文件中添加以下配置:

 
  1. <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.11 -->
  2. <dependency>
  3.     <groupId>org.apache.spark</groupId>
  4.     <artifactId>spark-core_2.11</artifactId>
  5.     <version>2.1.1</version>
  6.     <scope>runtime</scope>
  7. </dependency>

为了后面打包时把依赖包也一起打包,需要把<scope>provided</scope>配置成<scope>runtime</scope>

2、配置spark-streaming_2.11-2.1.1.jar

往pom.xml文件中添加以下配置:

 
  1.  <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming_2.11 -->
  2. <dependency>
  3.     <groupId>org.apache.spark</groupId>
  4.     <artifactId>spark-streaming_2.11</artifactId>
  5.     <version>2.1.1</version>
  6.     <scope>runtime</scope>
  7. </dependency>

 为了后面打包时把依赖包也一起打包,需要把<scope>provided</scope>配置成<scope>runtime</scope>。

三、配置Spark+Kafka

1、配置spark-streaming-kafka-0-8_2.11-2.1.1.jar

往pom.xml文件中添加以下配置:

 
  1. <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka-0-8_2.11 -->
  2. <dependency>
  3.     <groupId>org.apache.spark</groupId>
  4.     <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
  5.     <version>2.1.1</version>
  6. </dependency>

四、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/xsd/maven-4.0.0.xsd">
  3. <modelVersion>4.0.0</modelVersion>
  4.  
  5. <groupId>Test</groupId>
  6. <artifactId>test</artifactId>
  7. <version>0.0.1-SNAPSHOT</version>
  8. <packaging>jar</packaging>
  9.  
  10. <name>test</name>
  11. <url>http://maven.apache.org</url>
  12.  
  13. <properties>
  14. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  15. <encoding>UTF-8</encoding>
  16. <!-- 配置JDK为1.8 -->
  17.         <java.version>1.8</java.version>
  18.         <maven.compiler.source>1.8</maven.compiler.source>
  19.         <maven.compiler.target>1.8</maven.compiler.target>
  20. </properties>
  21.  
  22. <dependencies>
  23. <dependency>
  24. <groupId>junit</groupId>
  25. <artifactId>junit</artifactId>
  26. <version>3.8.1</version>
  27. <scope>test</scope>
  28. </dependency>
  29. <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.11 -->
  30. <dependency>
  31.             <groupId>org.apache.spark</groupId>
  32.             <artifactId>spark-core_2.11</artifactId>
  33.             <version>2.1.1</version>
  34.             <scope>runtime</scope>
  35. </dependency>
  36.  <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming_2.11 -->
  37. <dependency>
  38.             <groupId>org.apache.spark</groupId>
  39.             <artifactId>spark-streaming_2.11</artifactId>
  40.             <version>2.1.1</version>
  41.             <scope>runtime</scope>
  42. </dependency>
  43. <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka-0-8_2.11 -->
  44. <dependency>
  45.          <groupId>org.apache.spark</groupId>
  46.          <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
  47.          <version>2.1.1</version>
  48. </dependency>
  49. </dependencies>
  50.  
  51. <build>
  52. <plugins>
  53. <!-- 配置JDK为1.8 -->
  54. <plugin>  
  55.                 <groupId>org.apache.maven.plugins</groupId>  
  56.                 <artifactId>maven-compiler-plugin</artifactId>  
  57.                 <configuration>  
  58.                     <source>1.8</source>  
  59.                     <target>1.8</target>  
  60.                 </configuration>  
  61.             </plugin>  
  62.         
  63.              <!-- 配置打包依赖包maven-assembly-plugin -->
  64. <plugin>
  65. <artifactId> maven-assembly-plugin </artifactId>
  66. <configuration>
  67. <descriptorRefs>
  68. <descriptorRef>jar-with-dependencies</descriptorRef>
  69. </descriptorRefs>
  70. <archive>
  71. <manifest>
  72. <mainClass></mainClass>
  73. </manifest>
  74. </archive>
  75. </configuration>
  76. <executions>
  77. <execution>
  78. <id>make-assembly</id>
  79. <phase>package</phase>
  80. <goals>
  81. <goal>assembly</goal>
  82. </goals>
  83. </execution>
  84. </executions>
  85. </plugin>
  86. </plugins>
  87. </build>
  88. </project>

五、本地开发spark代码上传spark集群服务并运行

JavaDirectKafkaCompare.java

 
  1. package com.spark.main;
  2.  
  3. import java.util.HashMap;
  4. import java.util.HashSet;
  5. import java.util.Arrays;
  6. import java.util.Iterator;
  7. import java.util.Map;
  8. import java.util.Set;
  9. import java.util.regex.Pattern;
  10.  
  11. import scala.Tuple2;
  12. import kafka.serializer.StringDecoder;
  13.  
  14. import org.apache.spark.SparkConf;
  15. import org.apache.spark.api.java.function.*;
  16. import org.apache.spark.streaming.api.java.*;
  17. import org.apache.spark.streaming.kafka.KafkaUtils;
  18. import org.apache.spark.streaming.Durations;
  19.  
  20. public class JavaDirectKafkaCompare {
  21.  
  22. public static void main(String[] args) throws Exception {
  23. /**
  24.  * setMaster("local[2]"),至少要指定两个线程,一条用于用于接收消息,一条线程用于处理消息
  25.  *  Durations.seconds(2)每两秒读取一次kafka
  26.  */
  27.     SparkConf sparkConf = new SparkConf().setAppName("JavaDirectKafkaWordCount").setMaster("local[2]");
  28.     JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
  29.     /**
  30.      * checkpoint("hdfs://192.168.168.200:9000/checkpoint")防止数据丢包
  31.      */
  32.     jssc.checkpoint("hdfs://192.168.168.200:9000/checkpoint");
  33.     /**
  34.      * 配置连接kafka的相关参数      
  35.      */
  36.     Set<String> topicsSet = new HashSet<>(Arrays.asList("test"));
  37.     Map<String, String> kafkaParams = new HashMap<>();
  38.     kafkaParams.put("metadata.broker.list", "192.168.168.200:9092");
  39.  
  40.     // Create direct kafka stream with brokers and topics
  41.     JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
  42.         jssc,
  43.         String.class,
  44.         String.class,
  45.         StringDecoder.class,
  46.         StringDecoder.class,
  47.         kafkaParams,
  48.         topicsSet
  49.     );
  50.  
  51.     // Get the lines, split them into words, count the words and print
  52.     /**
  53.      * _2()获取第二个对象的值
  54.      */
  55.     JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
  56.       @Override
  57.       public String call(Tuple2<String, String> tuple2) {
  58.         return tuple2._2();
  59.       }
  60.     });
  61.     
  62.    String sfzh = "432922196105276721";
  63.    JavaDStream<String> wordCounts = lines.filter(new Function<String, Boolean>(){
  64. @Override
  65. public Boolean call(String s) throws Exception {
  66. // TODO Auto-generated method stub
  67. /**
  68.  * 通过身份证号筛选出相关数据
  69.  */
  70. if(s.contains(sfzh)){
  71. System.out.println("比对出来的结果:" + s);
  72. return true;
  73. }
  74. return false;
  75. }
  76.    });
  77.    wordCounts.print();
  78.     // Start the computation
  79.     jssc.start();
  80.     jssc.awaitTermination();
  81. }
  82.  
  83. }

右键Run As ------>Maven install,运行成功之后,会在target目录生成一个test-0.0.1-SNAPSHOT-jar-with-dependencies.jar,把该jar包复制到LInux集群环境下的SPARK_HOME/myApp目录下:

执行命令:

 
  1. cd /usr/local/spark/spark-2.1.1-bin-hadoop2.7;
  2. bin/spark-submit --class "com.spark.main.JavaDirectKafkaCompare" --master local[4] myApp/test-0.0.1-SNAPSHOT-jar-with-dependencies.jar;

六、附上离线Maven仓库

 下载地址:  链接:http://pan.baidu.com/s/1eS7Ywme 密码:y3qz

© 著作权归作者所有

共有 人打赏支持
四叶草666
粉丝 0
博文 51
码字总数 50778
作品 0
深圳
程序员
私信 提问
vSphere 5.5 VM整合磁盘失败之—文件被锁定无法访问

vSphere 5.5 VM整合磁盘失败之—文件被锁定无法访问 环境:vSPhere 5.5u3,虚机使用EMC的networker备份 问题现象:在vc上发现,晚上经过networker的备份之后,虚机提示需要整合磁盘 解决前相...

Makka_Pakka
07/06
0
0
springboot从入门到精通教程分享, 深度掌握Springboot实践技术教程

深度掌握Springboot实践技术教程,共22个课时,需要的可回复邮箱。 课程简介 day1 01springboot简介-曾经开发中的问题 02springboot解决的问题-优势以及弊端 03springboot环境准备 04spring...

小小倾听
12/06
0
0
白俊遥/thinkphp-bjyadmin

创建 QQ 群及捐赠渠道 链接 博客:http://baijunyao.com github:https://github.com/baijunyao/thinkphp-bjyadmin oschina:http://git.oschina.net/shuaibai123/thinkphp-bjyadmin 简介 使......

白俊遥
2016/06/30
0
0
玩转springboot:整合mybatis实例

这篇文章讲解一下springboot整合mybatis,其实,springboot整合mybatis和springmvc整合mybatis并没有什么太大的区别,大体上还是差不多哦,只是比springmvc更加的简单一点,下面我们就以一个...

java知识分子
10/30
0
0
Confluence 6 数据库整合的限制

数据库整合的限制 注意: Confluence 自带的 XML 方式导出方法并不适用于备份和整合大数据集。这里有一些第三方的数据库工具你可以使用能够帮助你对大数据集进行备份和整合。如果你在选择正确...

honeymose
06/05
0
0

没有更多内容

加载失败,请刷新页面

加载更多

deepin中配置robot framework环境

本文永久更新地址:https://my.oschina.net/bysu/blog/2989005 【若要到岸,请摇船:开源中国 不最醉不龟归】 1.在终端中输入pip,回车,如果提示没有该命令,则先安转pip sudo apt-get inst...

不最醉不龟归
14分钟前
1
0
OSChina 周日乱弹 —— 钱不还,我就当你人不在了

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @莱布妮子 :分享Bigleaf的单曲《小鹿》 《小鹿》- Bigleaf 手机党少年们想听歌,请使劲儿戳(这里) 周日在家做什么? 做手工呀, @poorfis...

小小编辑
今天
74
3
EOS docker开发环境

使用eos docker镜像是部署本地EOS开发环境的最轻松愉快的方法。使用官方提供的eos docker镜像,你可以快速建立一个eos开发环境,可以迅速启动开发节点和钱包服务器、创建账户、编写智能合约....

汇智网教程
今天
20
0
《唐史原来超有趣》的读后感优秀范文3700字

《唐史原来超有趣》的读后感优秀范文3700字: 作者:花若离。我今天分享的内容《唐史原来超有趣》这本书的读后感,我将这本书看了一遍之后就束之高阁了,不过里面的内容一直在在脑海中回放,...

原创小博客
今天
29
0
IC-CAD Methodology知识图谱

CAD (Computer Aided Design),计算机辅助设计,指利用计算机及其图形设备帮助设计人员进行设计工作,这个定义同样可以用来近似描述IC公司CAD工程师这个岗位的工作。 早期IC公司的CAD岗位最初...

李艳青1987
今天
33
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部