文档章节

Maven+Eclipse+SparkStreaming+Kafka整合

四叶草666
 四叶草666
发布于 2017/07/18 15:36
字数 1102
阅读 18
收藏 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
深圳
程序员
snakerflow/snaker-springmvc

Introduction Snaker-SpringMVC项目主要是基于springMVC、spring3、hibernate3、snaker框架整合的一个最基本的流程管理模块,方便大家轻松地完成流程引擎的整合 ###整合步骤 ####1).依赖包整...

snakerflow
2014/11/30
0
0
vSphere 5.5 VM整合磁盘失败之—文件被锁定无法访问

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

Makka_Pakka
07/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
struts + spring + hibernate 不太理解这种搭配,请指点一下。

struts + spring + hibernate 整合这是到底什么意思嘛?struts 是一个java的web开发框架,spring也是,为什么很多时候总是struts+spring ,是不是把这两个框架整合?如何整合开发? 使用php...

hstaewg
2015/08/29
181
5
ThinkPHP 常用功能和 SDK 合集--thinkphp-bjyadmin

简介 使用 thinkphp 开发项目的过程中把一些常用的功能或者第三方 sdk 整合好;开源供亲们参考; 这些都是经过线上运营考验的;无毒害可以免费放心折腾使用;只要不会某一天找到我说因为借鉴了...

白俊遥
2017/09/01
1K
2

没有更多内容

加载失败,请刷新页面

加载更多

20180920 rzsz传输文件、用户和用户组相关配置文件与管理

利用rz、sz实现Linux与Windows互传文件 [root@centos01 ~]# yum install -y lrzsz # 安装工具sz test.txt # 弹出对话框,传递到选择的路径下rz # 回车后,会从对话框中选择对应的文件传递...

野雪球
今天
2
0
OSChina 周四乱弹 —— 毒蛇当辣条

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @ 达尔文:分享花澤香菜/前野智昭/小野大輔/井上喜久子的单曲《ミッション! 健?康?第?イチ》 《ミッション! 健?康?第?イチ》- 花澤香菜/前野智...

小小编辑
今天
7
3
java -jar运行内存设置

java -Xms64m #JVM启动时的初始堆大小 -Xmx128m #最大堆大小 -Xmn64m #年轻代的大小,其余的空间是老年代 -XX:MaxMetaspaceSize=128m # -XX:CompressedClassSpaceSize=6...

李玉长
今天
4
0
Spring | 手把手教你SSM最优雅的整合方式

HEY 本节主要内容为:基于Spring从0到1搭建一个web工程,适合初学者,Java初级开发者。欢迎与我交流。 MODULE 新建一个Maven工程。 不论你是什么工具,选这个就可以了,然后next,直至finis...

冯文议
今天
2
0
RxJS的另外四种实现方式(四)——性能最高的库(续)

接上一篇RxJS的另外四种实现方式(三)——性能最高的库 上一篇文章我展示了这个最高性能库的实现方法。下面我介绍一下这个性能提升的秘密。 首先,为了弄清楚Most库究竟为何如此快,我必须借...

一个灰
今天
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部