针对单个源的具有多个接收器的结构化流

knpiaxh1  于 2021-05-27  发布在  Spark
关注(0)|答案(2)|浏览(475)

我们正在创建一个spark结构化流应用程序，其中有一个源和多个接收器（kafka和hdfs）。我们有Spark了 2.3.1 使用群集 foreachBatch 在 DataStreamWriter 我们够不着。因此，我选择了这里和其他许多地方提倡的多流编写器方法。不幸的是，该流中只有一个对所有批处理都有效，而另一个仍处于休眠状态。我缺少什么来解决我的问题？

val spark = SparkSession
      .builder
      .appName("Spark-Structured-Streaming")
      .enableHiveSupport()
      .getOrCreate()

  val lines = spark.readStream
    .format("socket")
    .option("host", "localhost")
    .option("port", 9999)
    .load()

  val words: Dataset[String] = lines.as[String]
    .flatMap(_.split(" "))
    .map(_.concat(Random.nextInt(10).toString))

  /**THIS DOESN'T STREAM TO HDFS (This work's only when the below one is commented)**/
  words.writeStream
    .format("parquet")
    .option("path", "path/to/destination/dir")
    .option("checkpointLocation", "some_location_1")
    .start()

  /**THIS WORK'S**/
  words.writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("topic", "updates")
    .option("checkpointLocation", "some_location_2")
    .start()

  spark.streams.awaitAnyTermination()

我只看到数据发布到Kafka，但我没有看到数据写入hdfs。只有当我评论出Kafka出版流时，沉睡的才有了生命。
更新：一个类似的易于复制的脚本，带有简单的sink

def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder.appName("Spark-Structured-Streaming").master("local[4]").getOrCreate()

    val lines = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()

    import spark.implicits._

    val words: Dataset[String] = lines.as[String].flatMap(_.split(" ")).map(_.concat(Random.nextInt(10).toString))

    words.writeStream.queryName("query1").format("console").start()

    words.writeStream.queryName("query2").foreach(new ForeachWriter[String] {
        override def open(partitionId: Long, version: Long): Boolean = true

        override def process(value: String): Unit = {
          println(value)
        }

        override def close(errorOrNull: Throwable): Unit = {}
      })
      .start()

    spark.streams.awaitAnyTermination()
  }

日志：输入 A B C ```
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
20/05/30 12:17:37 INFO SparkContext: Running Spark version 2.3.1
20/05/30 12:17:38 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
20/05/30 12:17:38 INFO SparkContext: Submitted application: Spark-Structured-Streaming
20/05/30 12:17:38 INFO SecurityManager: Changing view acls to: z001v7w
20/05/30 12:17:38 INFO SecurityManager: Changing modify acls to: z001v7w
20/05/30 12:17:38 INFO SecurityManager: Changing view acls groups to:
20/05/30 12:17:38 INFO SecurityManager: Changing modify acls groups to:
20/05/30 12:17:38 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(z001v7w); groups with view permissions: Set(); users with modify permissions: Set(z001v7w); groups with modify permissions: Set()
20/05/30 12:17:38 INFO Utils: Successfully started service 'sparkDriver' on port 49262.
20/05/30 12:17:39 INFO SparkEnv: Registering MapOutputTracker
20/05/30 12:17:39 INFO SparkEnv: Registering BlockManagerMaster
20/05/30 12:17:39 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
20/05/30 12:17:39 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
20/05/30 12:17:39 INFO DiskBlockManager: Created local directory at /private/var/folders/rr/vxr2k38j6_qcl3xmbf2g_25933gq0_/T/blockmgr-80dcbdff-eb1c-49e5-a33b-47a897735d31
20/05/30 12:17:39 INFO MemoryStore: MemoryStore started with capacity 2004.6 MB
20/05/30 12:17:39 INFO SparkEnv: Registering OutputCommitCoordinator
20/05/30 12:17:39 INFO Utils: Successfully started service 'SparkUI' on port 4040.
20/05/30 12:17:39 INFO SparkUI: Bound SparkUI to 127.0.0.1, and started at http://localhost:4040
20/05/30 12:17:39 INFO Executor: Starting executor ID driver on host localhost
20/05/30 12:17:39 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 49263.
20/05/30 12:17:39 INFO NettyBlockTransferService: Server created on localhost:49263
20/05/30 12:17:39 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
20/05/30 12:17:39 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, localhost, 49263, None)
20/05/30 12:17:39 INFO BlockManagerMasterEndpoint: Registering block manager localhost:49263 with 2004.6 MB RAM, BlockManagerId(driver, localhost, 49263, None)
20/05/30 12:17:39 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, localhost, 49263, None)
20/05/30 12:17:39 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, localhost, 49263, None)
20/05/30 12:17:40 INFO SharedState: Setting hive.metastore.warehouse.dir ('null') to the value of spark.sql.warehouse.dir ('file:/Users//spark-warehouse/').
20/05/30 12:17:40 INFO SharedState: Warehouse path is 'file://spark-warehouse/'.
20/05/30 12:17:40 INFO StateStoreCoordinatorRef: Registered StateStoreCoordinator endpoint
20/05/30 12:17:40 WARN TextSocketSourceProvider: The socket source should not be used for production applications! It does not support recovery.
20/05/30 12:17:43 INFO MicroBatchExecution: Starting query1 [id = a5714e40-553f-4d21-9115-9c0d9ffe950b, runId = 8d206486-36d8-4319-ad3d-fa5cb00a79ca]. Use file:///private/var/folders/rr/vxr2k38j6_qcl3xmbf2g_25933gq0_/T/temporary-5f818bb9-b347-432f-8c43-07fccec0e9e8 to store the query checkpoint.
20/05/30 12:17:43 WARN ClosureCleaner: Expected a closure; got com.tgt.sign.POCTestMain$$anon$1
20/05/30 12:17:43 INFO MicroBatchExecution: Starting new streaming query.
20/05/30 12:17:43 INFO MicroBatchExecution: Streaming query made progress: {
"id" : "a5714e40-553f-4d21-9115-9c0d9ffe950b",
"runId" : "8d206486-36d8-4319-ad3d-fa5cb00a79ca",
"name" : "query1",
"timestamp" : "2020-05-30T17:17:43.915Z",
"batchId" : 0,
"numInputRows" : 0,
"processedRowsPerSecond" : 0.0,
"durationMs" : {
"getOffset" : 0,
"triggerExecution" : 10
},
"stateOperators" : [ ],
"sources" : [ {
"description" : "TextSocketSource[host: localhost, port: 9999]",
"startOffset" : null,
"endOffset" : null,
"numInputRows" : 0,
"processedRowsPerSecond" : 0.0
} ],
"sink" : {
"description" : "org.apache.spark.sql.execution.streaming.ConsoleSinkProvider@5e761dda"
}
}
20/05/30 12:17:43 INFO MicroBatchExecution: Starting query2 [id = 776c6082-920a-4406-913f-c1b2eda5f12d, runId = 7d066ebb-fe7b-454c-98c5-2ed05a8c0bda]. Use file:///private/var/folders/rr/vxr2k38j6_qcl3xmbf2g_25933gq0_/T/temporary-7a3485d8-9973-43a7-8a35-4a5dd0745719 to store the query checkpoint.
20/05/30 12:17:43 INFO MicroBatchExecution: Starting new streaming query.
20/05/30 12:17:43 INFO MicroBatchExecution: Streaming query made progress: {
"id" : "776c6082-920a-4406-913f-c1b2eda5f12d",
"runId" : "7d066ebb-fe7b-454c-98c5-2ed05a8c0bda",
"name" : "query2",
"timestamp" : "2020-05-30T17:17:43.991Z",
"batchId" : 0,
"numInputRows" : 0,
"processedRowsPerSecond" : 0.0,
"durationMs" : {
"getOffset" : 0,
"triggerExecution" : 1
},
"stateOperators" : [ ],
"sources" : [ {
"description" : "TextSocketSource[host: localhost, port: 9999]",
"startOffset" : null,
"endOffset" : null,
"numInputRows" : 0,
"processedRowsPerSecond" : 0.0
} ],
"sink" : {
"description" : "ForeachSink"
}
}
20/05/30 12:17:49 INFO MicroBatchExecution: Committed offsets for batch 0. Metadata OffsetSeqMetadata(0,1590859069597,Map(spark.sql.shuffle.partitions -> 200, spark.sql.streaming.stateStore.providerClass -> org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider))
20/05/30 12:17:50 INFO CodeGenerator: Code generated in 474.062005 ms
20/05/30 12:17:51 INFO CodeGenerator: Code generated in 58.908865 ms
20/05/30 12:17:51 INFO WriteToDataSourceV2Exec: Start processing data source writer: org.apache.spark.sql.execution.streaming.sources.MicroBatchWriter@6ccc266. The input RDD has 4 partitions.
20/05/30 12:17:51 INFO SparkContext: Starting job: start at POCTestMain.scala:147
20/05/30 12:17:51 INFO DAGScheduler: Got job 0 (start at POCTestMain.scala:147) with 4 output partitions
20/05/30 12:17:51 INFO DAGScheduler: Final stage: ResultStage 0 (start at POCTestMain.scala:147)
20/05/30 12:17:51 INFO DAGScheduler: Parents of final stage: List()
20/05/30 12:17:51 INFO DAGScheduler: Missing parents: List()
20/05/30 12:17:51 INFO DAGScheduler: Submitting ResultStage 0 (MapPartitionsRDD[5] at start at POCTestMain.scala:147), which has no missing parents
20/05/30 12:17:51 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 12.0 KB, free 2004.6 MB)
20/05/30 12:17:51 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 5.3 KB, free 2004.6 MB)
20/05/30 12:17:51 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:49263 (size: 5.3 KB, free: 2004.6 MB)
20/05/30 12:17:51 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1039
20/05/30 12:17:51 INFO DAGScheduler: Submitting 4 missing tasks from ResultStage 0 (MapPartitionsRDD[5] at start at POCTestMain.scala:147) (first 15 tasks are for partitions Vector(0, 1, 2, 3))
20/05/30 12:17:51 INFO TaskSchedulerImpl: Adding task set 0.0 with 4 tasks
20/05/30 12:17:51 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, executor driver, partition 0, PROCESS_LOCAL, 7851 bytes)
20/05/30 12:17:51 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1, localhost, executor driver, partition 1, PROCESS_LOCAL, 7851 bytes)
20/05/30 12:17:51 INFO TaskSetManager: Starting task 2.0 in stage 0.0 (TID 2, localhost, executor driver, partition 2, PROCESS_LOCAL, 7851 bytes)
20/05/30 12:17:51 INFO TaskSetManager: Starting task 3.0 in stage 0.0 (TID 3, localhost, executor driver, partition 3, PROCESS_LOCAL, 7995 bytes)
20/05/30 12:17:51 INFO Executor: Running task 3.0 in stage 0.0 (TID 3)
20/05/30 12:17:51 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
20/05/30 12:17:51 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
20/05/30 12:17:51 INFO Executor: Running task 2.0 in stage 0.0 (TID 2)
20/05/30 12:17:51 INFO CodeGenerator: Code generated in 25.371124 ms
20/05/30 12:17:51 INFO CodeGenerator: Code generated in 12.870672 ms
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 0 is committing.
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 2 is committing.
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 1 is committing.
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 0 committed.
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 1 committed.
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 2 committed.
20/05/30 12:17:51 INFO CodeGenerator: Code generated in 14.157876 ms
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 3 is committing.
20/05/30 12:17:51 INFO DataWritingSparkTask: Writer for partition 3 committed.
20/05/30 12:17:51 INFO Executor: Finished task 2.0 in stage 0.0 (TID 2). 1240 bytes result sent to driver
20/05/30 12:17:51 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 1240 bytes result sent to driver
20/05/30 12:17:51 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 1240 bytes result sent to driver
20/05/30 12:17:51 INFO Executor: Finished task 3.0 in stage 0.0 (TID 3). 2297 bytes result sent to driver
20/05/30 12:17:51 INFO TaskSetManager: Finished task 2.0 in stage 0.0 (TID 2) in 227 ms on localhost (executor driver) (1/4)
20/05/30 12:17:51 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 256 ms on localhost (executor driver) (2/4)
20/05/30 12:17:51 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 233 ms on localhost (executor driver) (3/4)
20/05/30 12:17:51 INFO TaskSetManager: Finished task 3.0 in stage 0.0 (TID 3) in 232 ms on localhost (executor driver) (4/4)
20/05/30 12:17:51 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
20/05/30 12:17:51 INFO DAGScheduler: ResultStage 0 (start at POCTestMain.scala:147) finished in 0.557 s
20/05/30 12:17:51 INFO DAGScheduler: Job 0 finished: start at POCTestMain.scala:147, took 0.616436 s
20/05/30 12:17:51 INFO WriteToDataSourceV2Exec: Data source writer org.apache.spark.sql.execution.streaming.sources.MicroBatchWriter@6ccc266 is committing.

Batch: 0

+-----+
|value|
+-----+
| A8|
| B9|
| C4|
+-----+

hadoopapache-sparkapache-spark-sqlspark-structured-streamingspark-streaming

2条答案

按热度按时间
ee7vknir

ee7vknir1#

现在我明白你的问题了。您正在使用 socket 作为输入源&启动两个线程用于消费和写入接收器。
在 socket (netcat) 例如，一旦从套接字使用数据，它就不可用 query1 首先，它将从套接字读取数据并写入接收器。 query2 没有数据可从套接字读取，因此它将只是处于空闲状态。
在spark文件中提到- The socket source should not be used for production applications it does not support recovery 而不是 socket 尝试流式传输 files &检查是否按预期工作。

赞(0）分享  回复(0）举报  2021-05-27
mbskvtky

mbskvtky2#

我也面临同样的问题，把代码改成了

/**THIS DOESN'T STREAM TO HDFS (This work's only when the below one is commented)**/
  val hdfsPush = words.write
    .format("parquet")
    .outputMode("append")
    .option("path", "path/to/destination/dir")
    .option("checkpointLocation", "some_location_1")

  /**THIS WORK'S**/
  val kafkaPush = words.writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("topic", "updates")
    .option("checkpointLocation", "some_location_2")

kafkaPush.start()
hdfsPush.start().awaitTermination()
赞(0）分享  回复(0）举报  2021-05-27
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com