maven - 当来自 Kafka 的 Spark 流数据显示 Horton 工作中的连接超时异常时

标签 maven apache-spark apache-kafka spark-streaming hortonworks-sandbox

我想在 Horton 作品中使用来自 Kafka 主题的 spark 流式传输数据。我已经启动了 zookeeper 和 Kafka 服务器。然后我在 Eclipse 中成功构建了一个带有所有流依赖项的 maven 项目。下面给出的是 pom 文件。

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>test</groupId>
<artifactId>spark-kafka</artifactId>
<version>1.0-SNAPSHOT</version>
<repositories>
    <repository>
        <id>hortonworks</id>
        <name>hortonworks repo</name>
        <url>http://repo.hortonworks.com/content/repositories/releases/</url>
    </repository>
</repositories>
<dependencies>

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka_2.10</artifactId>
<version>1.3.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
<dependency>

<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.0</version>
<scope>provided</scope>
</dependency>

</dependencies>
<build>
    <defaultGoal>package</defaultGoal>
    <resources>
        <resource>
            <directory>src/main/resources</directory>
            <filtering>true</filtering>
        </resource>
        <resource>
            <directory>src/test/resources</directory>
            <filtering>true</filtering>
        </resource>
    </resources>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-resources-plugin</artifactId>
            <configuration>
                <encoding>UTF-8</encoding>
            </configuration>
            <executions>
                <execution>
                    <goals>
                        <goal>copy-resources</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.2.0</version>
            <configuration>
                <recompileMode>incremental</recompileMode>
                <args>
                    <arg>-target:jvm-1.7</arg>
                </args>
                <javacArgs>
                    <javacArg>-source</javacArg>
                    <javacArg>1.7</javacArg>
                    <javacArg>-target</javacArg>
                    <javacArg>1.7</javacArg>
                </javacArgs>
            </configuration>
            <executions>
                <execution>
                    <id>scala-compile</id>
                    <phase>process-resources</phase>
                    <goals>
                        <goal>compile</goal>
                    </goals>
                </execution>
                <execution>
                    <id>scala-test-compile</id>
                    <phase>process-test-resources</phase>
                    <goals>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>1.7</source>
                <target>1.7</target>
            </configuration>

            <executions>
                <execution>
                    <phase>compile</phase>
                    <goals>
                        <goal>compile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>


        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>2.3</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                </execution>
            </executions>
            <configuration>
                <filters>
                    <filter>
                        <artifact>*:*</artifact>
                        <excludes>
                            <exclude>META-INF/*.SF</exclude>
                            <exclude>META-INF/*.DSA</exclude>
                            <exclude>META-INF/*.RSA</exclude>
                        </excludes>
                    </filter>
                </filters>
                <finalName>uber-${project.artifactId}-${project.version}</finalName>
            </configuration>
        </plugin>

    </plugins>

</build>

流媒体.SCALA 
package com.jesperdj.example

import org.apache.spark._
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils
object HelloScala  {
def main(args:Array[String])
{
val conf = new SparkConf().setMaster("local[*]").setAppName("KafkaReceiver");
val ssc = new StreamingContext(conf, Seconds(10))

val kafkaStream = KafkaUtils.createStream(ssc, "localhost:2181","spark-group", Map("streaming" -> 5))
//need to change the topic name and the port number accordingly

kafkaStream.print()  //prints the stream of data received


ssc.start()
ssc.awaitTermination()
}
}

在 Eclipse 中运行 scala 程序时，它显示以下错误，
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
18/05/03 12:15:18 INFO SparkContext: Running Spark version 1.6.0
18/05/03 12:15:19 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
18/05/03 12:15:19 INFO SecurityManager: Changing view acls to: u60888
18/05/03 12:15:19 INFO SecurityManager: Changing modify acls to: u60888
18/05/03 12:15:19 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(u60888); users with modify permissions: Set(u60888)
18/05/03 12:15:20 INFO Utils: Successfully started service 'sparkDriver' on port 59787.
18/05/03 12:15:21 INFO Slf4jLogger: Slf4jLogger started
18/05/03 12:15:21 INFO Remoting: Starting remoting
18/05/03 12:15:22 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriverActorSystem@10.100.170.138:59800]
18/05/03 12:15:22 INFO Utils: Successfully started service 'sparkDriverActorSystem' on port 59800.
18/05/03 12:15:22 INFO SparkEnv: Registering MapOutputTracker
18/05/03 12:15:22 INFO SparkEnv: Registering BlockManagerMaster
18/05/03 12:15:22 INFO DiskBlockManager: Created local directory at C:\Users\u60888\AppData\Local\Temp\19\blockmgr-eda85d0d-70f0-48c8-8910-9a883bacdd38
18/05/03 12:15:22 INFO MemoryStore: MemoryStore started with capacity 4.4 GB
18/05/03 12:15:22 INFO SparkEnv: Registering OutputCommitCoordinator
18/05/03 12:15:23 INFO Utils: Successfully started service 'SparkUI' on port 4040.
18/05/03 12:15:23 INFO SparkUI: Started SparkUI at http://10.100.170.138:4040
18/05/03 12:15:23 INFO Executor: Starting executor ID driver on host localhost
18/05/03 12:15:23 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 59829.
18/05/03 12:15:23 INFO NettyBlockTransferService: Server created on 59829
18/05/03 12:15:23 INFO BlockManagerMaster: Trying to register BlockManager
18/05/03 12:15:23 INFO BlockManagerMasterEndpoint: Registering block manager localhost:59829 with 4.4 GB RAM, BlockManagerId(driver, localhost, 59829)
18/05/03 12:15:23 INFO BlockManagerMaster: Registered BlockManager
18/05/03 12:15:25 INFO ReceiverTracker: Starting 1 receivers
18/05/03 12:15:25 INFO ReceiverTracker: ReceiverTracker started
18/05/03 12:15:25 INFO ForEachDStream: metadataCleanupDelay = -1
18/05/03 12:15:25 INFO KafkaInputDStream: metadataCleanupDelay = -1
18/05/03 12:15:25 INFO KafkaInputDStream: Slide time = 10000 ms
18/05/03 12:15:25 INFO KafkaInputDStream: Storage level = StorageLevel(false, false, false, false, 1)
18/05/03 12:15:25 INFO KafkaInputDStream: Checkpoint interval = null
18/05/03 12:15:25 INFO KafkaInputDStream: Remember duration = 10000 ms
18/05/03 12:15:25 INFO KafkaInputDStream: Initialized and validated org.apache.spark.streaming.kafka.KafkaInputDStream@269e1b99
18/05/03 12:15:25 INFO ForEachDStream: Slide time = 10000 ms
18/05/03 12:15:25 INFO ForEachDStream: Storage level = StorageLevel(false, false, false, false, 1)
18/05/03 12:15:25 INFO ForEachDStream: Checkpoint interval = null
18/05/03 12:15:25 INFO ForEachDStream: Remember duration = 10000 ms
18/05/03 12:15:25 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream@7f09a1b9
18/05/03 12:15:25 INFO RecurringTimer: Started timer for JobGenerator at time 1525329930000
18/05/03 12:15:25 INFO JobGenerator: Started JobGenerator at 1525329930000 ms
18/05/03 12:15:25 INFO JobScheduler: Started JobScheduler
18/05/03 12:15:25 INFO StreamingContext: StreamingContext started
18/05/03 12:15:26 INFO ReceiverTracker: Receiver 0 started
18/05/03 12:15:26 INFO DAGScheduler: Got job 0 (start at HelloScala.scala:20) with 1 output partitions
18/05/03 12:15:26 INFO DAGScheduler: Final stage: ResultStage 0 (start at HelloScala.scala:20)
18/05/03 12:15:26 INFO DAGScheduler: Parents of final stage: List()
18/05/03 12:15:26 INFO DAGScheduler: Missing parents: List()
18/05/03 12:15:26 INFO DAGScheduler: Submitting ResultStage 0 (Receiver 0 ParallelCollectionRDD[0] at makeRDD at ReceiverTracker.scala:588), which has no missing parents
18/05/03 12:15:26 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 34.1 KB, free 34.1 KB)
18/05/03 12:15:26 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 11.0 KB, free 45.1 KB)
18/05/03 12:15:26 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:59829 (size: 11.0 KB, free: 4.4 GB)
18/05/03 12:15:26 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1006
18/05/03 12:15:26 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 0 (Receiver 0 ParallelCollectionRDD[0] at makeRDD at ReceiverTracker.scala:588)
18/05/03 12:15:26 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
18/05/03 12:15:26 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, partition 0,PROCESS_LOCAL, 2988 bytes)
18/05/03 12:15:26 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
18/05/03 12:15:26 INFO RecurringTimer: Started timer for BlockGenerator at time 1525329926800
18/05/03 12:15:26 INFO BlockGenerator: Started BlockGenerator
18/05/03 12:15:26 INFO BlockGenerator: Started block pushing thread
18/05/03 12:15:26 INFO ReceiverTracker: Registered receiver for stream 0 from 10.100.170.138:59787
18/05/03 12:15:26 INFO ReceiverSupervisorImpl: Starting receiver
18/05/03 12:15:26 INFO KafkaReceiver: Starting Kafka Consumer Stream with group: spark-group
18/05/03 12:15:26 INFO KafkaReceiver: Connecting to Zookeeper: localhost:2181
18/05/03 12:15:26 INFO VerifiableProperties: Verifying properties
18/05/03 12:15:26 INFO VerifiableProperties: Property group.id is overridden to spark-group
18/05/03 12:15:26 INFO VerifiableProperties: Property zookeeper.connect is overridden to localhost:2181
18/05/03 12:15:26 INFO VerifiableProperties: Property zookeeper.connection.timeout.ms is overridden to 10000
18/05/03 12:15:26 INFO ZookeeperConsumerConnector: [spark-group_trvhsdapv036-1525329926880-ccdd31c7], Connecting to zookeeper instance at localhost:2181
18/05/03 12:15:26 INFO ZkEventThread: Starting ZkClient event thread.
18/05/03 12:15:26 INFO ZooKeeper: Client environment:zookeeper.version=3.4.5-1392090, built on 09/30/2012 17:52 GMT
18/05/03 12:15:26 INFO ZooKeeper: Client environment:host.name=trvhsdapv036.ustr.com
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.version=1.8.0
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.vendor=Oracle Corporation
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.home=C:\Program Files\Java\jdk1.8.0\jre
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.class.path=U:\KafkaWorkspace2\scala-maven-example-master\scala-maven-example-master\target\classes;U:\KafkaWorkspace2\scala-maven-example-master\scala-maven-example-master\target\test-classes;U:\maven\org\apache\spark\spark-streaming-kafka_2.10\1.3.0\spark-streaming-kafka_2.10-1.3.0.jar;U:\maven\org\apache\kafka\kafka_2.10\0.8.1.1\kafka_2.10-0.8.1.1.jar;U:\maven\com\yammer\metrics\metrics-core\2.2.0\metrics-core-2.2.0.jar;U:\maven\org\xerial\snappy\snappy-java\1.0.5\snappy-java-1.0.5.jar;U:\maven\com\101tec\zkclient\0.3\zkclient-0.3.jar;U:\maven\org\spark-project\spark\unused\1.0.0\unused-1.0.0.jar;U:\maven\org\apache\spark\spark-streaming_2.10\1.6.0\spark-streaming_2.10-1.6.0.jar;U:\maven\org\apache\spark\spark-core_2.10\1.6.0\spark-core_2.10-1.6.0.jar;U:\maven\org\apache\avro\avro-mapred\1.7.7\avro-mapred-1.7.7-hadoop2.jar;U:\maven\org\apache\avro\avro-ipc\1.7.7\avro-ipc-1.7.7.jar;U:\maven\org\apache\avro\avro\1.7.7\avro-1.7.7.jar;U:\maven\org\apache\avro\avro-ipc\1.7.7\avro-ipc-1.7.7-tests.jar;U:\maven\org\codehaus\jackson\jackson-core-asl\1.9.13\jackson-core-asl-1.9.13.jar;U:\maven\org\codehaus\jackson\jackson-mapper-asl\1.9.13\jackson-mapper-asl-1.9.13.jar;U:\maven\com\twitter\chill_2.10\0.5.0\chill_2.10-0.5.0.jar;U:\maven\com\esotericsoftware\kryo\kryo\2.21\kryo-2.21.jar;U:\maven\com\esotericsoftware\reflectasm\reflectasm\1.07\reflectasm-1.07-shaded.jar;U:\maven\com\esotericsoftware\minlog\minlog\1.2\minlog-1.2.jar;U:\maven\org\objenesis\objenesis\1.2\objenesis-1.2.jar;U:\maven\com\twitter\chill-java\0.5.0\chill-java-0.5.0.jar;U:\maven\org\apache\xbean\xbean-asm5-shaded\4.4\xbean-asm5-shaded-4.4.jar;U:\maven\org\apache\hadoop\hadoop-client\2.2.0\hadoop-client-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-common\2.2.0\hadoop-common-2.2.0.jar;U:\maven\commons-cli\commons-cli\1.2\commons-cli-1.2.jar;U:\maven\org\apache\commons\commons-math\2.1\commons-math-2.1.jar;U:\maven\xmlenc\xmlenc\0.52\xmlenc-0.52.jar;U:\maven\commons-configuration\commons-configuration\1.6\commons-configuration-1.6.jar;U:\maven\commons-collections\commons-collections\3.2.1\commons-collections-3.2.1.jar;U:\maven\commons-digester\commons-digester\1.8\commons-digester-1.8.jar;U:\maven\commons-beanutils\commons-beanutils\1.7.0\commons-beanutils-1.7.0.jar;U:\maven\commons-beanutils\commons-beanutils-core\1.8.0\commons-beanutils-core-1.8.0.jar;U:\maven\org\apache\hadoop\hadoop-auth\2.2.0\hadoop-auth-2.2.0.jar;U:\maven\org\apache\commons\commons-compress\1.4.1\commons-compress-1.4.1.jar;U:\maven\org\tukaani\xz\1.0\xz-1.0.jar;U:\maven\org\apache\hadoop\hadoop-hdfs\2.2.0\hadoop-hdfs-2.2.0.jar;U:\maven\org\mortbay\jetty\jetty-util\6.1.26\jetty-util-6.1.26.jar;U:\maven\org\apache\hadoop\hadoop-mapreduce-client-app\2.2.0\hadoop-mapreduce-client-app-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-mapreduce-client-common\2.2.0\hadoop-mapreduce-client-common-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-yarn-client\2.2.0\hadoop-yarn-client-2.2.0.jar;U:\maven\com\google\inject\guice\3.0\guice-3.0.jar;U:\maven\javax\inject\javax.inject\1\javax.inject-1.jar;U:\maven\aopalliance\aopalliance\1.0\aopalliance-1.0.jar;U:\maven\com\sun\jersey\jersey-test-framework\jersey-test-framework-grizzly2\1.9\jersey-test-framework-grizzly2-1.9.jar;U:\maven\com\sun\jersey\jersey-test-framework\jersey-test-framework-core\1.9\jersey-test-framework-core-1.9.jar;U:\maven\javax\servlet\javax.servlet-api\3.0.1\javax.servlet-api-3.0.1.jar;U:\maven\com\sun\jersey\jersey-client\1.9\jersey-client-1.9.jar;U:\maven\com\sun\jersey\jersey-grizzly2\1.9\jersey-grizzly2-1.9.jar;U:\maven\org\glassfish\grizzly\grizzly-http\2.1.2\grizzly-http-2.1.2.jar;U:\maven\org\glassfish\grizzly\grizzly-framework\2.1.2\grizzly-framework-2.1.2.jar;U:\maven\org\glassfish\gmbal\gmbal-api-only\3.0.0-b023\gmbal-api-only-3.0.0-b023.jar;U:\maven\org\glassfish\external\management-api\3.0.0-b012\management-api-3.0.0-b012.jar;U:\maven\org\glassfish\grizzly\grizzly-http-server\2.1.2\grizzly-http-server-2.1.2.jar;U:\maven\org\glassfish\grizzly\grizzly-rcm\2.1.2\grizzly-rcm-2.1.2.jar;U:\maven\org\glassfish\grizzly\grizzly-http-servlet\2.1.2\grizzly-http-servlet-2.1.2.jar;U:\maven\org\glassfish\javax.servlet\3.1\javax.servlet-3.1.jar;U:\maven\com\sun\jersey\jersey-json\1.9\jersey-json-1.9.jar;U:\maven\org\codehaus\jettison\jettison\1.1\jettison-1.1.jar;U:\maven\stax\stax-api\1.0.1\stax-api-1.0.1.jar;U:\maven\com\sun\xml\bind\jaxb-impl\2.2.3-1\jaxb-impl-2.2.3-1.jar;U:\maven\javax\xml\bind\jaxb-api\2.2.2\jaxb-api-2.2.2.jar;U:\maven\javax\activation\activation\1.1\activation-1.1.jar;U:\maven\org\codehaus\jackson\jackson-jaxrs\1.8.3\jackson-jaxrs-1.8.3.jar;U:\maven\org\codehaus\jackson\jackson-xc\1.8.3\jackson-xc-1.8.3.jar;U:\maven\com\sun\jersey\contribs\jersey-guice\1.9\jersey-guice-1.9.jar;U:\maven\org\apache\hadoop\hadoop-yarn-server-common\2.2.0\hadoop-yarn-server-common-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-mapreduce-client-shuffle\2.2.0\hadoop-mapreduce-client-shuffle-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-yarn-api\2.2.0\hadoop-yarn-api-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-mapreduce-client-core\2.2.0\hadoop-mapreduce-client-core-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-yarn-common\2.2.0\hadoop-yarn-common-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-mapreduce-client-jobclient\2.2.0\hadoop-mapreduce-client-jobclient-2.2.0.jar;U:\maven\org\apache\hadoop\hadoop-annotations\2.2.0\hadoop-annotations-2.2.0.jar;U:\maven\org\apache\spark\spark-launcher_2.10\1.6.0\spark-launcher_2.10-1.6.0.jar;U:\maven\org\apache\spark\spark-network-common_2.10\1.6.0\spark-network-common_2.10-1.6.0.jar;U:\maven\org\apache\spark\spark-network-shuffle_2.10\1.6.0\spark-network-shuffle_2.10-1.6.0.jar;U:\maven\org\fusesource\leveldbjni\leveldbjni-all\1.8\leveldbjni-all-1.8.jar;U:\maven\com\fasterxml\jackson\core\jackson-annotations\2.4.4\jackson-annotations-2.4.4.jar;U:\maven\org\apache\spark\spark-unsafe_2.10\1.6.0\spark-unsafe_2.10-1.6.0.jar;U:\maven\net\java\dev\jets3t\jets3t\0.7.1\jets3t-0.7.1.jar;U:\maven\commons-codec\commons-codec\1.3\commons-codec-1.3.jar;U:\maven\commons-httpclient\commons-httpclient\3.1\commons-httpclient-3.1.jar;U:\maven\org\apache\curator\curator-recipes\2.4.0\curator-recipes-2.4.0.jar;U:\maven\org\apache\curator\curator-framework\2.4.0\curator-framework-2.4.0.jar;U:\maven\org\apache\curator\curator-client\2.4.0\curator-client-2.4.0.jar;U:\maven\org\apache\zookeeper\zookeeper\3.4.5\zookeeper-3.4.5.jar;U:\maven\jline\jline\0.9.94\jline-0.9.94.jar;U:\maven\com\google\guava\guava\14.0.1\guava-14.0.1.jar;U:\maven\org\eclipse\jetty\orbit\javax.servlet\3.0.0.v201112011016\javax.servlet-3.0.0.v201112011016.jar;U:\maven\org\apache\commons\commons-lang3\3.3.2\commons-lang3-3.3.2.jar;U:\maven\org\apache\commons\commons-math3\3.4.1\commons-math3-3.4.1.jar;U:\maven\com\google\code\findbugs\jsr305\1.3.9\jsr305-1.3.9.jar;U:\maven\org\slf4j\slf4j-api\1.7.10\slf4j-api-1.7.10.jar;U:\maven\org\slf4j\jul-to-slf4j\1.7.10\jul-to-slf4j-1.7.10.jar;U:\maven\org\slf4j\jcl-over-slf4j\1.7.10\jcl-over-slf4j-1.7.10.jar;U:\maven\log4j\log4j\1.2.17\log4j-1.2.17.jar;U:\maven\org\slf4j\slf4j-log4j12\1.7.10\slf4j-log4j12-1.7.10.jar;U:\maven\com\ning\compress-lzf\1.0.3\compress-lzf-1.0.3.jar;U:\maven\net\jpountz\lz4\lz4\1.3.0\lz4-1.3.0.jar;U:\maven\org\roaringbitmap\RoaringBitmap\0.5.11\RoaringBitmap-0.5.11.jar;U:\maven\commons-net\commons-net\2.2\commons-net-2.2.jar;U:\maven\com\typesafe\akka\akka-remote_2.10\2.3.11\akka-remote_2.10-2.3.11.jar;U:\maven\com\typesafe\akka\akka-actor_2.10\2.3.11\akka-actor_2.10-2.3.11.jar;U:\maven\com\typesafe\config\1.2.1\config-1.2.1.jar;U:\maven\io\netty\netty\3.8.0.Final\netty-3.8.0.Final.jar;U:\maven\com\google\protobuf\protobuf-java\2.5.0\protobuf-java-2.5.0.jar;U:\maven\org\uncommons\maths\uncommons-maths\1.2.2a\uncommons-maths-1.2.2a.jar;U:\maven\com\typesafe\akka\akka-slf4j_2.10\2.3.11\akka-slf4j_2.10-2.3.11.jar;U:\maven\org\json4s\json4s-jackson_2.10\3.2.10\json4s-jackson_2.10-3.2.10.jar;U:\maven\org\json4s\json4s-core_2.10\3.2.10\json4s-core_2.10-3.2.10.jar;U:\maven\org\json4s\json4s-ast_2.10\3.2.10\json4s-ast_2.10-3.2.10.jar;U:\maven\org\scala-lang\scalap\2.10.0\scalap-2.10.0.jar;U:\maven\org\scala-lang\scala-compiler\2.10.0\scala-compiler-2.10.0.jar;U:\maven\com\sun\jersey\jersey-server\1.9\jersey-server-1.9.jar;U:\maven\asm\asm\3.1\asm-3.1.jar;U:\maven\com\sun\jersey\jersey-core\1.9\jersey-core-1.9.jar;U:\maven\org\apache\mesos\mesos\0.21.1\mesos-0.21.1-shaded-protobuf.jar;U:\maven\io\netty\netty-all\4.0.29.Final\netty-all-4.0.29.Final.jar;U:\maven\com\clearspring\analytics\stream\2.7.0\stream-2.7.0.jar;U:\maven\io\dropwizard\metrics\metrics-core\3.1.2\metrics-core-3.1.2.jar;U:\maven\io\dropwizard\metrics\metrics-jvm\3.1.2\metrics-jvm-3.1.2.jar;U:\maven\io\dropwizard\metrics\metrics-json\3.1.2\metrics-json-3.1.2.jar;U:\maven\io\dropwizard\metrics\metrics-graphite\3.1.2\metrics-graphite-3.1.2.jar;U:\maven\com\fasterxml\jackson\core\jackson-databind\2.4.4\jackson-databind-2.4.4.jar;U:\maven\com\fasterxml\jackson\core\jackson-core\2.4.4\jackson-core-2.4.4.jar;U:\maven\com\fasterxml\jackson\module\jackson-module-scala_2.10\2.4.4\jackson-module-scala_2.10-2.4.4.jar;U:\maven\org\scala-lang\scala-reflect\2.10.4\scala-reflect-2.10.4.jar;U:\maven\com\thoughtworks\paranamer\paranamer\2.6\paranamer-2.6.jar;U:\maven\org\apache\ivy\ivy\2.4.0\ivy-2.4.0.jar;U:\maven\oro\oro\2.0.8\oro-2.0.8.jar;U:\maven\org\tachyonproject\tachyon-client\0.8.2\tachyon-client-0.8.2.jar;U:\maven\commons-lang\commons-lang\2.4\commons-lang-2.4.jar;U:\maven\commons-io\commons-io\2.4\commons-io-2.4.jar;U:\maven\org\tachyonproject\tachyon-underfs-hdfs\0.8.2\tachyon-underfs-hdfs-0.8.2.jar;U:\maven\org\tachyonproject\tachyon-underfs-s3\0.8.2\tachyon-underfs-s3-0.8.2.jar;U:\maven\org\tachyonproject\tachyon-underfs-local\0.8.2\tachyon-underfs-local-0.8.2.jar;U:\maven\net\razorvine\pyrolite\4.9\pyrolite-4.9.jar;U:\maven\net\sf\py4j\py4j\0.9\py4j-0.9.jar
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.library.path=C:\Program Files\Java\jdk1.8.0\bin;C:\Windows\Sun\Java\bin;C:\Windows\system32;C:\Windows;C:\ProgramData\Oracle\Java\javapath;C:\Windows\system32;C:\Windows;C:\Windows\System32\Wbem;C:\Windows\System32\WindowsPowerShell\v1.0\;C:\Program Files\Citrix\Virtual Desktop Agent\;C:\Program Files (x86)\Common Files\Citrix\System32;C:\Program Files\nodejs\;C:\Program Files (x86)\Citrix\system32;C:\Program Files (x86)\Google\Chrome\Application;C:\ProgramData\Oracle\Java\javapath;C:\Windows\system32;C:\Windows;C:\Windows\System32\Wbem;C:\Windows\System32\WindowsPowerShell\v1.0\;C:\Program Files (x86)\Citrix\system32;C:\Program Files\Citrix\Virtual Desktop Agent\;C:\Program Files (x86)\Common Files\Citrix\System32;U:\AppData\Roaming\Programs\Git\cmd;C:\Program Files\Java\jre1.8.0_31;C:\Program Files\Java\jdk1.8.0_31;.
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.io.tmpdir=C:\Users\u60888\AppData\Local\Temp\19\
18/05/03 12:15:26 INFO ZooKeeper: Client environment:java.compiler=<NA>
18/05/03 12:15:26 INFO ZooKeeper: Client environment:os.name=Windows Server 2008 R2
18/05/03 12:15:26 INFO ZooKeeper: Client environment:os.arch=amd64
18/05/03 12:15:26 INFO ZooKeeper: Client environment:os.version=6.1
18/05/03 12:15:26 INFO ZooKeeper: Client environment:user.name=u60888
18/05/03 12:15:26 INFO ZooKeeper: Client environment:user.home=C:\Users\u60888
18/05/03 12:15:26 INFO ZooKeeper: Client environment:user.dir=U:\KafkaWorkspace2\scala-maven-example-master\scala-maven-example-master
18/05/03 12:15:26 INFO ZooKeeper: Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@396c8257
18/05/03 12:15:26 INFO ClientCnxn: Opening socket connection to server 0:0:0:0:0:0:0:1/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)
18/05/03 12:15:27 WARN ClientCnxn: Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:712)
    at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:350)
    at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1068)
18/05/03 12:15:28 INFO ClientCnxn: Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
18/05/03 12:15:29 WARN ClientCnxn: Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:712)
    at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:350)
    at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1068)
18/05/03 12:15:30 INFO JobScheduler: Added jobs for time 1525329930000 ms
18/05/03 12:15:30 INFO JobScheduler: Starting job streaming job 1525329930000 ms.0 from job set of time 1525329930000 ms
-------------------------------------------
Time: 1525329930000 ms
-------------------------------------------

最佳答案

在pom文件中添加这些依赖

<dependencies>
  <dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>$2.10.4</version>
  </dependency>


  <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka -->
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka_2.10</artifactId>
    <version>1.6.2</version>
  </dependency>

  <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.8.2.2</version>
  </dependency>


  <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.2</version>
  </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.10</artifactId>
        <version>1.6.2</version>

    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>1.6.2</version>

    </dependency>

</dependencies>

试试这个代码，
//spark configurations
        val sparkConf = new SparkConf().setAppName("Kafka").setMaster("local[*]")
        val ssc = new StreamingContext(sparkConf, Seconds(10))
        val kafkaConf = Map(
        "metadata.broker.list" -> "localhost:9092",
        "zookeeper.connect" -> "localhost:9091",
        "group.id" -> "kafka-streaming-example",
        "zookeeper.connection.timeout.ms" -> "200000"
        )
        val lines = KafkaUtils.createStream[Array[Byte], String, DefaultDecoder, StringDecoder](
        ssc,
        kafkaConf,
        Map("sample-topic" -> 1), // subscripe to topic and partition 1
        StorageLevel.MEMORY_ONLY
        )
        println("printing" + lines.toString())
        val words = lines.flatMap { case (x, y) => y.split(" ") }
        words.print()
        ssc.start()
        ssc.awaitTermination()

希望这有效!

关于maven - 当来自 Kafka 的 Spark 流数据显示 Horton 工作中的连接超时异常时，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/50148591/

上一篇：unity3d - 如何避免 Unity 编译所有标准着色器变体

下一篇：php - 预期响应代码 354，但得到代码 "503"

相关文章：

dataframe - 如何在某些情况下在 Spark DataFrame 中创建新列 'count'

apache-spark - GCP 上的 PySpark PandasUDF - 内存分配

java - 如何根据输入 ID 在架构注册表中注册架构

maven - Jetty注解超时原因

apache-spark - 如何使用Spark上下文访问Web URL

java - 在 Artifactory 存储库中访问 JAR 文件失败

java - 使用 Kafka-Spark Streaming API 处理流数据时进行重复

apache-spark - Spark Streaming中如何将压缩数据写入Kafka？

spring - 在 Spring 中使用 LTW，无需使用 -javaagent

maven - Atmosphere ，jboss 和 primefaces 插入。让他们工作

©2024 IT工具网  联系我们