为什么我的spark工作停留在Kafka流媒体上

Question

spark作业提交到minicube创建的kubernetes集群中的spark集群后的输出：

----------------- RUNNING ----------------------
[Stage 0:>                                                          (0 + 0) / 2]17/06/16 16:08:15 INFO VerifiableProperties: Verifying properties
17/06/16 16:08:15 INFO VerifiableProperties: Property group.id is overridden to xxx
17/06/16 16:08:15 INFO VerifiableProperties: Property zookeeper.connect is overridden to 
xxxxxxxxxxxxxxxxxxxxx
[Stage 0:>                                                          (0 + 0) / 2]

来自spark web ui的信息：

我的文件中的foreachRDD。scala:49详细信息

组织。阿帕奇。火花流动。数据流。数据流。foreachRDD（DStream.scala:625）myfile。运行（myfile.scala:49）Myjob$。main（Myjob.scala:100）Myjob。main（Myjob.scala）sun。反映NativeMethodAccessorImpl。invoke0（本机方法）sun。反映NativeMethodAccessorImpl。调用（NativeMethodAccessorImpl.java:62）sun。反映DelegatingMethodAccessorImpl。调用（DelegatingMethodAccessorImpl.java:43）java。朗。反思。方法调用（Method.java:498）org。阿帕奇。火花部署SparkSubmit$。org$apache$spark$deploy$SparkSubmit$$runMain（SparkSubmit.scala:743）org。阿帕奇。火花部署SparkSubmit$。doRunMain$1（SparkSubmit.scala:187）org。阿帕奇。火花部署SparkSubmit$。提交（SparkSubmit.scala:212）org。阿帕奇。火花部署SparkSubmit$。main（SparkSubmit.scala:126）org。阿帕奇。火花部署SparkSubmit。main（SparkSubmit.scala）

我的代码：

  println("----------------- RUNNING ----------------------");
    eventsStream.foreachRDD { rdd =>
        println("xxxxxxxxxxxxxxxxxxxxx")
        //println(rdd.count());
    if( !rdd.isEmpty )
    {
      println("yyyyyyyyyyyyyyyyyyyyyyy")
        val df = sqlContext.read.json(rdd);
        df.registerTempTable("data");

        val rules = rulesSource.rules();
        var resultsRDD : RDD[(String,String,Long,Long,Long,Long,Long,Long)]= sc.emptyRDD;
        rules.foreach { rule =>
        ...
        }

        sqlContext.dropTempTable("data")
    }
    else
    {
        println("-------");
        println("NO DATA");
        println("-------");
    }
}

有什么想法吗？谢谢

更新

我的spark作业在独立spark的docker容器中运行良好。但若提交到kubernetes集群中的spark集群，它将被卡在Kafka流中。不知道为什么？

spark master的yaml文件来自https://github.com/phatak-dev/kubernetes-spark/blob/master/spark-master.yaml

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  labels:
    name: spark-master
  name: spark-master
spec:
  replicas: 1
  template:
    metadata:
      labels:
        name: spark-master
    spec:
      containers:
      - name : spark-master
        image: spark-2.1.0-bin-hadoop2.6 
        imagePullPolicy: "IfNotPresent"
        name: spark-master
        ports:
        - containerPort: 7077
          protocol: TCP
        command:
         - "/bin/bash"
         - "-c"
         - "--"
        args :
- './start-master.sh ; sleep infinity'

Answer 1

日志将有助于诊断问题。

本质上，你不能在RDD操作中创建另一个RDD。即rdd1.map{rdd2.count（）}无效

查看导入隐式sqlContext后RDD如何转换为数据帧。

        import sqlContext.implicits._
        eventsStream.foreachRDD { rdd =>

            println("yyyyyyyyyyyyyyyyyyyyyyy")

            val df = rdd.toDF(); 
            df.registerTempTable("data");
            .... //Your logic here.
            sqlContext.dropTempTable("data")
        }