0

在我们的 Spark 应用程序中，我们使用Spark structured streaming. 它使用Kafka as input stream, &HiveAcid as writeStream到 Hive 表。对于，它是从以下HiveAcid位置调用的开源库：https ://github.com/qubole/spark-acidspark acidqubole

下面是我们的代码：

import za.co.absa.abris.avro.functions.from_confluent_avro
....

val spark = SparkSession
  .builder()
  .appName("events")
  .config("spark.sql.streaming.metricsEnabled", true)
  .enableHiveSupport()
  .getOrCreate()

import spark.implicits._

val input_stream_df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka:9092")
  .option("startingOffsets", '{"events":{"0":2310384922,"1":2280420020,"2":2278027233,"3":2283047819,"4":2285647440}}')
  .option("maxOffsetsPerTrigger", 10000)
  .option("subscribe", "events")
  .load()

// schema registry config
val srConfig = Map(
  "schema.registry.url"           -> "http://schema-registry:8081",
  "value.schema.naming.strategy"  -> "topic.name",
  "schema.registry.topic"         -> "events",
  "value.schema.id"               -> "latest"
)

val data = input_stream_df
  .withColumn("value", from_confluent_avro(col("value"), srConfig))
  .withColumn("timestamp_s", from_unixtime($"value.timestamp" / 1000))
  .select(
    $"value.*",
    year($"timestamp_s")       as 'year,
    month($"timestamp_s")      as 'month,
    dayofmonth($"timestamp_s") as 'day
  )

// format "HiveAcid" is provided by spark-acid lib from Qubole
val output_stream_df = data.writeStream.format("HiveAcid")
  .queryName("hiveSink")
  .option("database", "default")
  .option("table", "events_sink")
  .option("checkpointLocation", "/user/spark/events/checkpoint")
  .option("spark.acid.streaming.log.metadataDir", "/user/spark/events/checkpoint/spark-acid")
  .option("metastoreUri", "thrift://hive-metastore:9083")
  .trigger(Trigger.ProcessingTime("30 seconds"))
  .start()

output_stream_df.awaitTermination()

我们能够将应用程序部署到生产环境，并重新部署了几次（~ 10 次）而没有问题。然后它遇到了以下错误：

查询 hiveSink [id = 080a9f25-23d2-4ec8-a8c0-1634398d6d29, runId = 990d3bba-0f7f-4bae-9f41-b43db6d1aeb3] 异常终止：作业因阶段失败而中止：阶段 0.0 中的任务 3 失败 4 次，最近一次失败：在 0.0 阶段丢失任务 3.3（TID 42、10.236.7.228、执行程序 3）：org.apache.hadoop.fs.FileAlreadyExistsException：/warehouse/tablespace/managed/hive/events/year=2020/month=5/day=客户端 10.236.7.228 的 18/delta_0020079_0020079/bucket_00003 已经存在 (...) 在 com.qubole.shaded.orc.impl.PhysicalFsWriter.(PhysicalFsWriter.java:95) 在 com.qubole.shaded.orc.impl.WriterImpl。 (WriterImpl.java:177) 在 com.qubole.shaded.hadoop.hive.ql.io.orc.WriterImpl.(WriterImpl.java:94) 在 com.qubole.shaded.hadoop.hive.ql.io.orc。 OrcFile.createWriter(OrcFile.java:334) 在 com.qubole.shaded.hadoop.hive.ql.io.orc。OrcRecordUpdater.initWriter(OrcRecordUpdater.java:602) 在 com.qubole.shaded.hadoop.hive.ql.io.orc.OrcRecordUpdater.addSimpleEvent(OrcRecordUpdater.java:423) 在 com.qubole.shaded.hadoop.hive.ql。 io.orc.OrcRecordUpdater.addSplitUpdateEvent(OrcRecordUpdater.java:432) 在 com.qubole.shaded.hadoop.hive.ql.io.orc.OrcRecordUpdater.insert(OrcRecordUpdater.java:484) 在 com.qubole.spark.hiveacid。 writer.hive.HiveAcidFullAcidWriter.process(HiveAcidWriter.scala:295) 在 com.qubole.spark.hiveacid.writer.TableWriter$$anon$1$$anonfun$6.apply(TableWriter.scala:153) 在 com.qubole.spark。 hiveacid.writer.TableWriter$$anon$1$$anonfun$6.apply(TableWriter.scala:153) (...) at com.qubole.spark.hiveacid.writer.TableWriter$$anon$1.apply(TableWriter.scala: 153）在 com.qubole.spark.hiveacid.writer.TableWriter$$anon$1。应用（TableWriter.scala:139）

每次重新启动应用程序时，都会显示不同的delta + bucket files已存在错误。但是，这些文件每次启动时都是新创建的（很可能），但不知道为什么会抛出错误。

任何指针将不胜感激。

4

1 回答 1

0

我从工人的错误日志中发现了真正的根本原因。这是由于我在使用的一个库中所做的代码更改导致了out of memory问题。

我之前发布的是来自驱动程序的错误日志，在工作节点上发生了几次故障之后。

于 2020-06-10T16:31:25.067 回答