2. 程式人生 > >Flume 、Kafka 與SparkStreaming 整合程式設計

Flume 、Kafka 與SparkStreaming 整合程式設計

阿新 • • 發佈：2019-01-01
Flume 、Kafka 與SparkStreaming 整合程式設計

一、 Kafka 與SparkStreaming 整合程式設計
1、程式
pull方式，可靠Recerver ，工作常用
com.imooc.spark . FlumePullWordCount .scala

package com.imooc.spark
import
 org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.flume.FlumeUtils
/** * 可靠Recerver ，工作常用 */ object
 FlumePullWordCount { def main(args: Array[ String ]): Unit = { if (args.length != 2 ) { System. err .println( "Usage: FlumePushWordCount <hostname> <port>" ) System.
exit ( 1 ) } val Array (hostname, port) = args val sparkConf = new SparkConf() //.setMaster("local[2]").setAppName("FlumePullWordCount") val ssc = new StreamingContext(sparkConf, Seconds ( 5 ))
val flumeStreame=FlumeUtils. createPollingStream (ssc,hostname, port.toInt) flumeStreame.map(x=> new String(x.event.getBody.array()).trim).flatMap(_.split( " " )).map((_, 1 )).reduceByKey(_+_).print()
ssc.start() ssc.awaitTermination() } }

push方式
com.imooc.spark . FlumePushWordCount .scala
package com.imooc.spark
import org.apache.spark.SparkConf import org.apache.spark.streaming.flume.FlumeUtils import org.apache.spark.streaming.{Seconds, StreamingContext}
object FlumePushWordCount { def main(args: Array[ String ]): Unit = { if (args.length != 2 ) { System.err.println("Usage: FlumePushWordCount <hostname> <port>") System.exit(1) } val Array (hostname, port) = args val sparkConf = new SparkConf() //.setMaster("local[2]").setAppName("FlumePushWordCount") val ssc = new StreamingContext(sparkConf, Seconds ( 5 ))
val flumeStreame=FlumeUtils. createStream (ssc,hostname, port.toInt) flumeStreame.map(x=> new String(x. event .getBody.array()).trim).flatMap(_.split( " " )).map((_, 1 )).reduceByKey(_+_).print()
ssc.start() ssc.awaitTermination() } }

pom.xml檔案
<project xmlns=" http://maven.apache.org/POM/4.0.0 " xmlns:xsi=" http://www.w3.org/2001/XMLSchema-instance "          xsi:schemaLocation=" http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd ">     <modelVersion>4.0.0</modelVersion>     <groupId>com.imooc.spark</groupId>     <artifactId>sparktrain</artifactId>     <version>1.0</version>     <inceptionYear>2008</inceptionYear>     <properties>         <scala.version>2.11.8</scala.version>         <kafka.version>0.9.0.0</kafka.version>         <spark.version>2.2.0</spark.version>         <hadoop.version>2.6.0-cdh5.7.0</hadoop.version>         <hbase.version>1.2.0-cdh5.7.0</hbase.version>     </properties>
    <!--新增cloudera的repository-->     <repositories>         <repository>             <id>cloudera</id>             <url> https://repository.cloudera.com/artifactory/cloudera-repos </url>         </repository>     </repositories>
    <dependencies>         <dependency>             <groupId>org.scala-lang</groupId>             <artifactId>scala-library</artifactId>             <version>${scala.version}</version>         </dependency>
        <!-- Kafka 依賴-->         <!--         <dependency>             <groupId>org.apache.kafka</groupId>             <artifactId>kafka_2.11</artifactId>             <version>${kafka.version}</version>         </dependency>         -->
        <!-- Hadoop 依賴-->         <dependency>             <groupId>org.apache.hadoop</groupId>             <artifactId>hadoop-client</artifactId>             <version>${hadoop.version}</version>         </dependency>
        <!-- HBase 依賴-->         <dependency>             <groupId>org.apache.hbase</groupId>             <artifactId>hbase-client</artifactId>             <version>${hbase.version}</version>         </dependency>
        <dependency>             <groupId>org.apache.hbase</groupId>             <artifactId>hbase-server</artifactId>             <version>${hbase.version}</version>         </dependency>
        <!-- Spark Streaming 依賴-->         <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming_2.11</artifactId>             <version>${spark.version}</version>         </dependency>

        <!-- Spark Streaming整合Flume 依賴-->         <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming-flume_2.11</artifactId>             <version>${spark.version}</version>         </dependency>
        <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming-flume-sink_2.11</artifactId>             <version>${spark.version}</version>         </dependency>
        <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>             <version>${spark.version}</version>         </dependency>
        <dependency>             <groupId> org.apache.commons </groupId>             <artifactId>commons-lang3</artifactId>             <version>3.5</version>         </dependency>
        <!-- Spark SQL 依賴-->         <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-sql_2.11</artifactId>             <version>${spark.version}</version>         </dependency>

        <dependency>             <groupId>com.fasterxml.jackson.module</groupId>             <artifactId>jackson-module-scala_2.11</artifactId>             <version>2.6.5</version>         </dependency>
        <dependency>             <groupId>net.jpountz.lz4</groupId>             <artifactId>lz4</artifactId>             <version>1.3.0</version>         </dependency>
        <dependency>              <groupId>mysql</groupId>             <artifactId>mysql-connector-java</artifactId>             <version>5.1.38</version>         </dependency>
        <dependency>             <groupId>org.apache.flume.flume-ng-clients</groupId>             <artifactId>flume-ng-log4jappender</artifactId>             <version>1.6.0</version>         </dependency>
    </dependencies>
    <build>         <sourceDirectory>src/main/scala</sourceDirectory>         <testSourceDirectory>src/test/scala</testSourceDirectory>         <plugins>             <plugin>                 <groupId>org.scala-tools</groupId>                 <artifactId>maven-scala-plugin</artifactId>                 <executions>                     <execution>                         <goals>                             <goal>compile</goal>                             <goal>testCompile</goal>                         </goals>                     </execution>                 </executions>                 <configuration>                     <scalaVersion>${scala.version}</scalaVersion>                     <args>                         <arg>-target:jvm-1.5</arg>                     </args>                 </configuration>             </plugin>             <plugin>                 <groupId>org.apache.maven.plugins</groupId>                 <artifactId>maven-eclipse-plugin</artifactId>                 <configuration>                     <downloadSources>true</downloadSources>                     <buildcommands>                         <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>                     </buildcommands>                     <additionalProjectnatures>                         <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>                     </additionalProjectnatures>                     <classpathContainers>                         <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>                         <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>                     </classpathContainers>                 </configuration>             </plugin>         </plugins>     </build>     <reporting>         <plugins>             <plugin>                 <groupId>org.scala-tools</groupId>                 <artifactId>maven-scala-plugin</artifactId>                 <configuration>                     <scalaVersion>${scala.version}</scalaVersion>                 </configuration>             </plugin>         </plugins>     </reporting> </project>

2、部署
1）、kafka部署   
 啟動kafka : kafka-server-start .sh $KAFKA_HOME/config /server .properties  建立topic ： kafka-topics .sh --create --zookeeper hadoop000 :2181 --replication-factor 1 --partitions 1 --topic test    生產者         ：   kafka-console-producer.sh --broker-list   hadoop000 :9092 --topic test
2）、提交作業(非聯網環境，不用packages ,而是用jars)
spark-submit \ --class com.imooc.spark. KafkaDirectWordCount \ --master local[2] \ --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0    \ #  --jars  spark-streaming-kafka-0-8-assembly had oop000 :9092 test

/www/lib/sparktrain-1.0.jar \ hadoop000 414 二、 Kafka 與SparkStreaming 整合程式設計
1、程式
com.imooc.spark.KafkaDirectWordCount.scala
package com.imooc.spark

import org.apache.spark.SparkConf import org.apache.spark.streaming.kafka.KafkaUtils import org.apache.spark.streaming.{Seconds, StreamingContext}
//Spark Streaming對接Kafka的方式二
object KafkaDirectWordCount {
def main(args: Array[String]): Unit = {
if (args.length != 2) { System. err .println( "Usage: KafkaDirectWordCount <brokers> <topics>" ) System. exit (1) }
val Array (brokers, topics) = args
val sparkConf = new SparkConf() //.setAppName("KafkaReceiverWordCount") //.setMaster("local[2]")
val ssc = new StreamingContext(sparkConf, Seconds (5))
val topicsSet = topics.split( "," ).toSet val kafkaParams = Map [String,String]( "metadata.broker.list" -> brokers)
// TODO... Spark Streaming如何對接Kafka val messages = KafkaUtils. createDirectStream [String,String,StringDecoder,StringDecoder]( ssc,kafkaParams,topicsSet )
// TODO... 自己去測試為什麼要取第二個 messages.map(_._2).flatMap(_.split( " " )).map((_,1)).reduceByKey(_+_).print()
ssc.start() ssc.awaitTermination() } }
pom.xml檔案
<project xmlns=" http://maven.apache.org/POM/4.0.0 " xmlns:xsi=" http://www.w3.org/2001/XMLSchema-instance "          xsi:schemaLocation=" http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd ">     <modelVersion>4.0.0</modelVersion>     <groupId>com.imooc.spark</groupId>     <artifactId>sparktrain</artifactId>     <version>1.0</version>     <inceptionYear>2008</inceptionYear>     <properties>         <scala.version>2.11.8</scala.version>         <kafka.version>0.9.0.0</kafka.version>         <spark.version>2.2.0</spark.version>         <hadoop.version>2.6.0-cdh5.7.0</hadoop.version>         <hbase.version>1.2.0-cdh5.7.0</hbase.version>     </properties>
    <!--新增cloudera的repository-->     <repositories>         <repository>             <id>cloudera</id>             <url> https://repository.cloudera.com/artifactory/cloudera-repos </url>         </repository>     </repositories>
    <dependencies>         <dependency>             <groupId>org.scala-lang</groupId>             <artifactId>scala-library</artifactId>             <version>${scala.version}</version>         </dependency>
        <!-- Kafka 依賴-->         <!--         <dependency>             <groupId>org.apache.kafka</groupId>             <artifactId>kafka_2.11</artifactId>             <version>${kafka.version}</version>         </dependency>         -->
        <!-- Hadoop 依賴-->         <dependency>             <groupId>org.apache.hadoop</groupId>             <artifactId>hadoop-client</artifactId>             <version>${hadoop.version}</version>         </dependency>
        <!-- HBase 依賴-->         <dependency>             <groupId>org.apache.hbase</groupId>             <artifactId>hbase-client</artifactId>             <version>${hbase.version}</version>         </dependency>
        <dependency>             <groupId>org.apache.hbase</groupId>             <artifactId>hbase-server</artifactId>             <version>${hbase.version}</version>         </dependency>
        <!-- Spark Streaming 依賴-->         <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming_2.11</artifactId>             <version>${spark.version}</version>         </dependency>

        <!-- Spark Streaming整合Flume 依賴-->         <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming-flume_2.11</artifactId>             <version>${spark.version}</version>         </dependency>
        <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming-flume-sink_2.11</artifactId>             <version>${spark.version}</version>         </dependency>
        <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>             <version>${spark.version}</version>         </dependency>
        <dependency>             <groupId> org.apache.commons </groupId>             <artifactId>commons-lang3</artifactId>             <version>3.5</version>         </dependency>
        <!-- Spark SQL 依賴-->         <dependency>             <groupId>org.apache.spark</groupId>             <artifactId>spark-sql_2.11</artifactId>             <version>${spark.version}</version>         </dependency>

        <dependency>             <groupId>com.fasterxml.jackson.module</groupId>             <artifactId>jackson-module-scala_2.11</artifactId>             <version>2.6.5</version>         </dependency>
        <dependency>             <groupId>net.jpountz.lz4</groupId>             <artifactId>lz4</artifactId>             <version>1.3.0</version>         </dependency>
        <dependency>              <groupId>mysql</groupId>             <artifactId>mysql-connector-java</artifactId>             <version>5.1.38</version>         </dependency>
        <dependency>             <groupId>org.apache.flume.flume-ng-clients</groupId>             <artifactId>flume-ng-log4jappender</artifactId>             <version>1.6.0</version>         </dependency>
    </dependencies>
    <build>         <sourceDirectory>src/main/scala</sourceDirectory>         <testSourceDirectory>src/test/scala</testSourceDirectory>         <plugins>             <plugin>                 <groupId>org.scala-tools</groupId>                 <artifactId>maven-scala-plugin</artifactId>                 <executions>                     <execution>                         <goals>                             <goal>compile</goal>                             <goal>testCompile</goal>                         </goals>                     </execution>                 </executions>                 <configuration>                     <scalaVersion>${scala.version}</scalaVersion>                     <args>                         <arg>-target:jvm-1.5</arg>                     </args>                 </configuration>             </plugin>             <plugin>                 <groupId>org.apache.maven.plugins</groupId>                 <artifactId>maven-eclipse-plugin</artifactId>                 <configuration>                     <downloadSources>true</downloadSources>                     <buildcommands>                         <buildcommand>ch.epfl.lamp.sdt.core.scalabuilder</buildcommand>                     </buildcommands>                     <additionalProjectnatures>                         <projectnature>ch.epfl.lamp.sdt.core.scalanature</projectnature>                     </additionalProjectnatures>                     <classpathContainers>                         <classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>                         <classpathContainer>ch.epfl.lamp.sdt.launching.SCALA_CONTAINER</classpathContainer>                     </classpathContainers>                 </configuration>             </plugin>         </plugins>     </build>

相關推薦

Flume 、Kafka 與SparkStreaming 整合程式設計

Flume 、Kafka 與SparkStreaming 整合程式設計 一、 Kafka 與SparkStreaming 整合程式設計 1、程式 pull方式，可靠Recerver ，工作常

Flume 、 Kafka 和SparkStreaming 簡單整合

flume 傳遞資料給Kafka ，然後Spark 從Kafka 中接收資料進行處理. 本文使用netcat 工具作為flume 的輸入源 , 話不多說，直接貼程式碼. 1、flume 配置檔案配置： a1.sources = r1 a1.sin

Flume、Kafka與Storm實現日誌處理

1. ZooKeeper 安裝參考 2. Kafka 2.1 解壓安裝 # 確保scala已經安裝好，本文安裝的是2.11.7 tar -xf kafka_2.11-0.9.0.1.tgz cd kafka_2.11-0.9.0.1 mkdi

大數據入門第二十四天——SparkStreaming（2）與flume、kafka整合

RM ESS 依賴 mep sock flume-ng bject 整合 master 前一篇中數據源采用的是從一個socket中拿數據，有點屬於“旁門左道”，正經的是從kafka等消息隊列中拿數據！ 主要支持的source，由官網得知如下：

kafka生產者消費者API 與sparkStreaming 整合（scala版）

maven配置檔案 <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka --> <dependency>

【SpringBoot、Gradle與Activti整合及入門HelloWord】

最近公司有一個工作流的需求，本來打算全部由業務實現工作流的全過程，後來為了簡化開發，選擇考慮使用工作流框架。後來考慮使用Activti和flowable,但是由於flowable網上的教程較少，最後選擇使用Activti。有興趣的同學可以去研究研究flowable。 使用技術：SpringBoo

kafka與storm整合

kafka與storm的整合步驟 採用官方storm-kafka-client方式進行整合 一 引入pom依賴 <properties> <project.build.sourceEncoding>UTF-8</

Java進階學習第二十四天（Spring框架：事務管理、Spring與Hibernate整合）

一、事務控制 1、引入 使用者訪問 > Action > Service > Dao 如何保證： 在service中呼叫2次dao，其中一個dao執行失敗，整個操作要回滾 2、事務控制概述 ① 程式設計式事務控制：自己手動控制事務 Jdbc程式

Kafka與SparkStreaming互動Direct之HB

在之前介紹了Kafka與SparkStreaming互動的兩種方式，我提到了公司採用的是Direct方式，這次我向大家分享一下將偏移量儲存在HBase中。 程式碼如下： package kafka1 import kafka.common.TopicAndPar

SODBASE CEP學習進階篇（二）續：日誌採集-Logstash、Kafka和CEP整合

相比Flume,筆者更推薦使用Logstash做日誌採集，見SODBASE CEP學習進階篇（二）續：日誌採集-Logstash、Kafka和CEP整合。如果之前專案中已經選型使用Flume，則本文供參考。 1. 啟動CEP模型 啟動CEP Server ./catalina

ActiveMQ訊息中介軟體 原理詳解 &&附demo實現、以及與springboot整合的demo

一、 訊息中介軟體概述1. 什麼是訊息中介軟體？ 面向訊息的中介軟體(MessageOrlented MiddlewareMOM)較好的解決了以上問題。傳送者將訊息傳送給訊息伺服器， 訊息伺服器將消感

HBase權威指南學習記錄（五、hbase與MapReduce整合）

新增依賴： <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifact

JAVAWEB開發之mybatis詳解(二)——高階對映、查詢快取、mybatis與Spring整合以及懶載入的配置和逆向工程

<?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE generatorConfiguration PUBLIC "-//mybatis.org//DTD MyBatis Generator Configuration 1.0//EN" "ht

kafka與flume的整合例項

一、整體步驟： 1.首先安裝kafka，配置flume。建立kafka的topic（利用zookeeper進行管理，所以首先要安裝zookeeper） 2.將檔案放置在flume的source目錄下，啟動flume。將檔案讀取到指定的kafka的topic中。 3.啟動的kafka的co

Centons7下安裝配置Flume、Linux下安裝配置Flume、Flume的簡單使用示例、Flume整合Kafka使用

本篇文章主要介紹在windows下使用vmware虛擬機器中的Linux(Centons7)下配置安裝Flume。 目前Flume 的最新版本為1.8，筆者安裝的是1.6，是Flume的一個經典版本，通常在生產環境中使用的就是這個版本，在安裝Flume前，看先它的前置條件  (1)

flume讀取binlog與kafka整合

一、現將kafka調通   檢視zookeeper的topic cd /usr/software/zookeeper/zookeeper/bin ./zkCli.sh start ls /brokers/topics   先來說一下，刪除kafka

Flume+HBase+Kafka整合與開發

 　　今天的內容是完成Flume+HBase+Kafka的整合開發。如下圖紅框中所示，節點1的Flume的source有兩個：節點2和節點3的sink輸出。節點1接收後進行預處理然後分別以AsyncHBaseSink（HBaseSink）和Kafka Sink的方式推送給

hbase與flume整合程式設計

1、官網下載src包，解壓，需要匯入的——》flume-ng-sinks——》flume-ng-hbase-sink 2、編輯SimpleAsyncHbaseEventSerializer：複製一份重新命名為MySimpleAsyncHbaseEventSerializer

kafka與flume的連線、kafka的python開發

kafka與flume的連線： 修改flume的配置檔案，將sinks的配置修改為： 看flume的官網： http://flume.apache.org/FlumeUserGuide.html#kafka-sink #定義gent的source channel sinksa5.sour

Flume+Kafka+SparkStreaming整合

目錄 1. Flume介紹 Flume是Cloudera提供的一個分散式、可靠、和高可用的海量日誌採集、聚合和傳輸的日誌收集系統，支援在日誌系統中定製各類資料傳送方，用於收集資料;同時，Flume提供對資料進行簡單處理，並寫到各種資料接受