在 Docker 上运行的 Zeppelin 中使用 Spark 使用 Kafka

Question

所以我有一个使用 Spotify/kafka 运行的本地 Docker 容器（在 Windows 10 上运行）。 对于这个容器，有一个名为“数据”的主题中发布的数据。

我可以像这样使用 Eclipse 中运行的 Java 应用程序中的数据：

private String topicName;
    private ConsumerConfig config;

    public KafkaConsumer() throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "test");
        props.put("client.id", this.getClass().getSimpleName());
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("partition.assignment.strategy", "range");

        this.config = new ConsumerConfig(props);
        this.topicName = "data";
    }

    public void run() {
        ConsumerConnector connector = Consumer.createJavaConsumerConnector(config);
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = connector.createMessageStreams(ImmutableMap.of(topicName, 1));
        List<KafkaStream<byte[], byte[]>> streams = messageStreams.get(topicName);
        ExecutorService executor = Executors.newFixedThreadPool(streams.size());

        for ( final KafkaStream<byte[], byte[]> stream : streams) {
            executor.submit(new Runnable() {
                public void run() {
                    ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
                    while (iterator.hasNext()) {
                        MessageAndMetadata<byte[], byte[]> messageAndMetadata = iterator.next();
                        String m = new String(messageAndMetadata.message());

                        System.out.println(m);
                    }
                }
            });
        }
    }

我现在想要的是在 dylanmei/zeppelin docker-container 中使用相同的主题。我尝试在 Zeppelin Notebook 中运行它：

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe


val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "localhost:9092",
  "zookeeper.connect" -> "localhost:2181",
  "group.id" -> "example",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "partition.assignment.strategy" -> "range",
)

val topics = Array("data")
val stream = KafkaUtils.createDirectStream[String, String](
  sc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

stream.map(record => (record.key, record.value))

我从spark kafka integration guide 中获取的内容

但我得到的只是连接被拒绝（当我插入错误的端口时也会发生这种情况，所以什么也没告诉我）：

java.net.ConnectException: Connection refused
at java.net.PlainSocketImpl.socketConnect(Native Method)
at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
at java.net.Socket.connect(Socket.java:589)
at org.apache.thrift.transport.TSocket.open(TSocket.java:182)
at org.apache.zeppelin.interpreter.remote.ClientFactory.create(ClientFactory.java:51)
at org.apache.zeppelin.interpreter.remote.ClientFactory.create(ClientFactory.java:37)
at org.apache.commons.pool2.BasePooledObjectFactory.makeObject(BasePooledObjectFactory.java:60)
at org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:861)
at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:435)
at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:363)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess.getClient(RemoteInterpreterProcess.java:96)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.init(RemoteInterpreter.java:216)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.getFormType(RemoteInterpreter.java:385)
at org.apache.zeppelin.interpreter.LazyOpenInterpreter.getFormType(LazyOpenInterpreter.java:105)
at org.apache.zeppelin.notebook.Paragraph.jobRun(Paragraph.java:306)
at org.apache.zeppelin.scheduler.Job.run(Job.java:176)
at org.apache.zeppelin.scheduler.RemoteScheduler$JobRunner.run(RemoteScheduler.java:329)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

希望有人能帮我让它运行，我不知道。

谢谢你和亲切的问候！

Answer 1

我认为这应该是火花解释器的问题。我遇到了同样的问题，当我重新启动我的解释器时一切顺利。您不妨检查一下解释器的配置。当您的解释器配置不正确时，有时您会收到该错误。