SparkException: задача не выполнена при записи строк - PullRequest
Новая
       33

SparkException: задача не выполнена при записи строк

0 голосов
/ 30 ноября 2018

Использование 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 Ответ

0 голосов
/ 30 ноября 2018

Вам необходимо создать новый сеанс улья, а затем использовать его для хранения данных в формате ORC.Код будет выглядеть так (непроверено, поскольку у меня нет доступа к любому искровому кластеру): 

def main(args: Array[String]): Unit = {
val spark = SparkSession
  .builder.master("local[4]")
  .appName("SpeedTester")
  .config("spark.driver.memory", "3g")
  .getOrCreate()

// create a new hive context from the spark context
val hiveContext = new org.apache.spark.sql.hive.HiveContext(spark)


// create the data frame and write it to orc
// output will be a directory of orc files
val ds = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "192.168.95.20:9092")
  .option("subscribe", "trial")
  .option("startingOffsets" , "earliest")
  .load()

ds.write.mode(SaveMode.Overwrite)
  .format("orc")
  .save("hdfs://192.168.95.19:8022/user/hive/warehouse/test.db/demo/")
}

Попробуйте и дайте мне знать, работает он или нет !!

...