Spark-Streaming連線kafka0.8 能連線卻不能消費問題
阿新 • • 發佈：2019-01-07
話不多說,直接上程式碼
package cn.sparkstreaming.kafka import kafka.serializer.{StringDecoder, Decoder} import org.apache.spark.streaming.kafka.KafkaUtils import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.{SparkContext, SparkConf} import scala.reflect.ClassTag /** * Created by Administrator on 2018/11/22. */ object SparkStreamDirectDemo { def main(args: Array[String]) { val conf = new SparkConf() conf.setAppName("spark_streaming") //conf.setMaster("local[*]") val sc = new SparkContext(conf) sc.setCheckpointDir("file:///segment2/Alarm_data/checkpoints") //sc.setCheckpointDir("checkpoints") sc.setLogLevel("ERROR") //多少秒消費一次 val ssc = new StreamingContext(sc, Seconds(60)) val topics = Map("WXALARM" -> 2) val kafkaParams = Map[String, String]( "bootstrap.servers" -> "10.216.5.152:9093,10.216.5.153:9093,10.216.5.154:9093", "group.id" -> "WYWX_123", "auto.offset.reset" -> "smallest" ) // 直連方式拉取資料，這種方式不會修改資料的偏移量，需要手動的更新 val lines = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set("WXALARM")).map(_._2) //val lines = KafkaUtils.createStream(ssc, "10.216.5.152:2183,10.216.5.153:2183,10.216.5.154:2183", "WYWX", topics).map(_._2) //lines.print() lines.saveAsTextFiles("file:///segment2/Alarm_data/Direct.txt") ssc.start() ssc.awaitTermination() } }
pom.xml
<?xml version="1.0" encoding="UTF-8"?> <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/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>esb_kafka</groupId> <artifactId>esb_kafka</artifactId> <version>1.0-SNAPSHOT</version> <properties> <maven.compiler.source>1.7</maven.compiler.source> <maven.compiler.target>1.7</maven.compiler.target> <encoding>UTF-8</encoding> <scala.version>2.11.8</scala.version> <spark.version>2.3.0</spark.version> </properties> <dependencies> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.4</version> <scope>test</scope> </dependency> <dependency> <groupId>org.specs</groupId> <artifactId>specs</artifactId> <version>1.2.5</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.6.0</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-8_2.11</artifactId> <version>${spark.version}</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.6</version> </dependency> <dependency> <groupId>org.apache.hive</groupId> <artifactId>hive-jdbc</artifactId> <version>0.13.0</version> </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.4.1</version> </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpcore</artifactId> <version>4.4.1</version> </dependency> <dependency> <groupId>io.spray</groupId> <artifactId>spray-json_2.10</artifactId> <version>1.3.2</version> </dependency> <dependency> <groupId>postgresql</groupId> <artifactId>postgresql</artifactId> <version>9.1-901.jdbc4</version> </dependency> <dependency> <groupId>c3p0</groupId> <artifactId>c3p0</artifactId> <version>0.9.1.2</version> </dependency> <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-ftp</artifactId> <version>2.13.2</version> </dependency> </dependencies> <build> <sourceDirectory>src/main/scala</sourceDirectory> <testSourceDirectory>src/test/scala</testSourceDirectory> <plugins> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> <executions> <execution> <goals> <goal>compile</goal> <goal>testCompile</goal> </goals> <configuration> <args> <arg>-dependencyfile</arg> <arg>${project.build.directory}/.scala_dependencies</arg> </args> </configuration> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.4.3</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build> </project>
因為在自己的叢集上是可以消費資料的,但是放在生產環境上不能消費,可能是kafka版本太低,也可能是cdh版本太高,導致不能相容
生產環境異常:
18/11/21 17:44:07 INFO spark.SparkContext: Running Spark version 2.3.0.cloudera4
18/11/21 17:44:07 INFO spark.SparkContext: Submitted application: spark_streaming
18/11/21 17:44:07 INFO spark.SecurityManager: Changing view acls to: root
18/11/21 17:44:07 INFO spark.SecurityManager: Changing modify acls to: root
18/11/21 17:44:07 INFO spark.SecurityManager: Changing view acls groups to:
18/11/21 17:44:07 INFO spark.SecurityManager: Changing modify acls groups to:
18/11/21 17:44:07 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(root); groups with view permissions: Set(); users with modify permissions: Set(root); groups with modify permissions: Set()
18/11/21 17:44:08 INFO util.Utils: Successfully started service 'sparkDriver' on port 7355.
18/11/21 17:44:08 INFO spark.SparkEnv: Registering MapOutputTracker
18/11/21 17:44:08 INFO spark.SparkEnv: Registering BlockManagerMaster
18/11/21 17:44:08 INFO storage.BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
18/11/21 17:44:08 INFO storage.BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
18/11/21 17:44:08 INFO storage.DiskBlockManager: Created local directory at /tmp/blockmgr-78ed315e-aa29-40b1-a44c-b1e05b0aea67
18/11/21 17:44:08 INFO memory.MemoryStore: MemoryStore started with capacity 366.3 MB
18/11/21 17:44:08 INFO spark.SparkEnv: Registering OutputCommitCoordinator
18/11/21 17:44:08 INFO util.log: Logging initialized @2256ms
18/11/21 17:44:08 INFO server.Server: jetty-9.3.z-SNAPSHOT, build timestamp: unknown, git hash: unknown
18/11/21 17:44:08 INFO server.Server: Started @2371ms
18/11/21 17:44:08 INFO server.AbstractConnector: Started 找了許久也沒有找到對應的解決方法,考慮到cdh的spark包和原生態的spark包存在一定的差異,就去官方看了一下
果不其然
需要把 spark-streaming-kafka-0-8-assembly_2.11.jar 放到spark的jars目錄下
因為我們cdh裝的最新spark版本2.3.0(cdh暫時沒有spark2.4.0版本)
官方又推薦的是spark 2.4.0版本,我發現 無論是2.4.0還是2.3.0的spark 都是可以執行的
groupId = org.apache.spark
artifactId = spark-streaming-kafka-0-8_2.11
version = 2.4.0希望我遇到的這個坑,能讓你解決當前問題