无法对 Kinesis 输入 Dstream 执行转换

Question

我正在尝试使用 kinesis 流并使用 spark 流对其执行转换，但没有得到结果。

现在 if 子句中的以下代码可以正常工作。当我在流中获得非零记录时，它会打印计数。 如果我在打印语句的情况下在 if 子句中使用 logger.warn("")，我只会继续在我的日志中获取以下行而不是实际计数：

INFO BlockManagerInfo：在 ip-10-225-172-215.ec2.internal:37820 的内存中添加了 input-0-1547121901568

不知道为什么 spark 在使用 logger 时执行 count() 失败。

到目前为止，我还可以将 str rdd 保存到 hdfs。 我的流正在以 json 格式输出记录。

现在，如果我尝试将我的流解析为 json 并像下面的代码一样加载，我会再次在我的日志中得到以下信息：

INFO BlockManagerInfo：在 ip-10-225-172-70.ec2.internal:39438 的内存中添加了 input-0-1547121901577（大小：6.4 MB，免费：1238.7 MB） 信息 BlockManagerInfo：在 ip-10-225-172-215.ec2.internal:37820 的内存中添加了 input-0-1547121901578（大小：5.1 MB，免费：1233.6 MB） INFO BlockManagerInfo：在 ip-10-225-172-70.ec2.internal:39438 的内存中添加了 input-0-1547121901578（大小：5.1 MB，免费：1233.6 MB） INFO JobScheduler：添加的作业时间为 1547121755000 毫秒

这是我正在使用的代码：

object KinesisWatch {


  val logger: Logger = Logger.getLogger("##### Kinesis-Logs #####")
  Logger.getLogger("org.apache.spark").setLevel(Level.INFO)
  Logger.getLogger("org.apache.spark.storage.BlockManager").setLevel(Level.INFO)
  logger.setLevel(Level.INFO)

  // Kinesis, AWS and Spark streaming configurations
  val kinesisConfig = Map(
    ("appName", "spark-kinesis"),
    ("streamName", "xxxx"),
    ("endpointUrl", "https://kinesis.us-east-1.amazonaws.com"),
    ("numStreams", "2"),
    ("regionName", "us-east-1"),
    ("awsAccessKey", "xxxxx"),
    ("awsSecretKey", "xxxxxx"),
    ("sparkStreamingBatchInterval", "5"),
    ("kinesisCheckpointInterval", "5"))

  logger.warn("Setting up SparkConfig and StreamingContext")
  val sparkSession = new SparkSessionWrapper().sparkSession
  val sc = sparkSession.sparkContext
  val sqlContext = sparkSession.sqlContext
  val ssc = new StreamingContext(sc, Seconds(kinesisConfig("sparkStreamingBatchInterval").toInt))

  val awsCredentials = SparkAWSCredentials.builder.basicCredentials(kinesisConfig("awsAccessKey"), kinesisConfig("awsSecretKey"))

  val kinesisStreams = (0 until kinesisConfig("numStreams").toInt).map { i =>
    KinesisInputDStream.builder
      .streamingContext(ssc)
      .endpointUrl(kinesisConfig("endpointUrl"))
      .regionName(kinesisConfig("regionName"))
      .streamName(kinesisConfig("streamName"))
      .initialPositionInStream(InitialPositionInStream.LATEST)
      .checkpointAppName(kinesisConfig("appName"))
      .checkpointInterval(Seconds(kinesisConfig("kinesisCheckpointInterval").toInt))
      .storageLevel(StorageLevel.MEMORY_AND_DISK_2)
      .kinesisCredentials(awsCredentials.build())
      .build()
  }
  val unionStreams = ssc.union(kinesisStreams)

  unionStreams.foreachRDD(rdd => {
    if
    (rdd.count()>0){
      println("New records found\nmetrics count in the batch: %s".format(rdd.count()))
      println("performing transformations")
      val str = rdd.map(str => new String(str))
      str.saveAsTextFile("/home/aman/EDA_BP_Kinesis")    
    }
    else
      println("No new record found")
  })    
  ssc.start()
  ssc.awaitTermination()

}

不工作：

 unionStreams.foreachRDD(rdd => {
    if
    (rdd.count()>0){
      println("New records found\nmetrics count in the batch: %s".format(rdd.count()))
      println("performing transformations")
      val str = rdd.map(str => new String(str))
      str.saveAsTextFile("/home/aman/EDA_BP_Kinesis")

      import sparkSession.implicits._
      val records = str.toString()
      val df = sparkSession.read.json(Seq(records).toDS)
      df.count()
      df.show()
    }
    else
      println("No new record found")
  })

我将分片数设为 2，流数设为 2，核心数设为 4。

Answer 1

尝试在 foreach 中创建单独的 Spark 会话。 希望这能解决您的死锁问题。

 unionStreams.foreachRDD(rdd => {
    if (rdd.count() > 0) {
      println("New records found\nmetrics count in the batch: %s".format(rdd.count()))
      println("performing transformations")
      val str = rdd.map(str => new String(str))
      str.saveAsTextFile("/home/aman/EDA_BP_Kinesis")
      val spark = SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()

      import spark.implicits._
      val records = str.toString()
      val df = spark.read.json(Seq(records).toDS)
      df.count()
      df.show()
    } else
      println("No new record found")})

Answer 2

你可以尝试增加工人的数量吗？在使用 Amazon Kinesis 进行 Spark Streaming 的情况下，我相信每个分片都需要一个专门的工作人员。所以如果你没有足够的工人，你的代码就不能工作。

为什么写入 hdfs 会成功？我猜文件写入操作不需要随机播放，因此每个分片读取工作者自己都可以处理它。但是对于count() 操作，这需要一些洗牌，因此需要更多的工人。