使用 spark-streaming
使用数据的步骤 Kafka
然后写信给 HDFS
在 orc
格式。
数据存储在 Kafka
就像:
hadoop
hive
impala
hive
我的代码:
def main(args: Array[String]): Unit = {
val spark = SparkSession
.builder.master("local[4]")
.appName("SpeedTester")
.config("spark.driver.memory", "3g")
.getOrCreate()
val ds = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "192.168.95.20:9092")
.option("subscribe", "trial")
.option("startingOffsets" , "earliest")
.load()
.selectExpr("CAST(value as string)")
.writeStream
.outputMode("append")
.format("orc")
.option("path", "hdfs://192.168.95.19:8022/user/hive/warehouse/test.db/demo")
.option("checkpointLocation", "/tmp/checkpoint")
.start()
.awaitTermination()
}
这些代码可以成功地写入数据 text
格式化为 HDFS
. 但是,当我把它变成 orc
格式,则返回:
Caused by: org.apache.spark.SparkException: Task failed while writing rows.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:285)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:197)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:196)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:381)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.FileNotFoundException: File does not exist: hdfs://192.168.95.19:8022/user/hive/warehouse/test.db/demo/part-00000-cfd9991f-e503-4140-811b-a00f7da7191e-c000.snappy.orc
at org.apache.hadoop.hdfs.DistributedFileSystem$20.doCall(DistributedFileSystem.java:1270)
at org.apache.hadoop.hdfs.DistributedFileSystem$20.doCall(DistributedFileSystem.java:1262)
这个问题的原因是什么?如何解决?感谢您的帮助。
顺便说一下 Hive
表格创建句子:
create table test.demo (demo string)
stored as orc;
1条答案
按热度按时间cfh9epnr1#
您需要创建一个新的配置单元会话,然后使用它以orc格式存储数据。代码看起来像(未经测试,因为我无法访问任何spark cluster):
试试这个,让我知道它是否有效!!