azure - 使用 Databricks 从 Azure EventHubs Capture 生成的 Azure Data Lake Gen1 读取 avro 数据失败

Question

我正在尝试从 Azure Data Lake Gen1 读取 avro 数据，这些数据是从 Azure EventHub 生成的，并且在 Azure Databricks 中使用 pyspark 启用了 Azure 事件中心捕获:

inputdata = "evenhubscapturepath/*/*"
rawData = spark.read.format("avro").load(inputdata)

以下语句失败

rawData.count()

与

org.apache.spark.SparkException: Job aborted due to stage failure: Task 162 in stage 48.0 failed 4 times, most recent failure: Lost task 162.3 in stage 48.0 (TID 2807, 10.3.2.4, executor 1): java.io.IOException: Not an Avro data file

EventHub-Capture 是否正在写入非 Avro 数据？是否有使用 Spark 读取 EventHub 捕获的数据的最佳实践？

Answer 1

实现冷摄取路径的一种模式是使用 Event Hubs Capture 。 EventHubs 捕获按照windowing parameters 的定义为每个分区写入一个文件。 。数据以 avro 格式写入，可以使用 Apache Spark 进行分析。

那么使用此功能的最佳实践是什么？

<强>1。不要过度分区

我经常看到人们使用默认配置，最终常常会产生许多小文件。如果您想使用 Spark 使用通过 EventHubs Capture 摄取的数据，请记住 file sizes in Azure Data Lake Store 的最佳实践。和 partitions与 Spark 。文件大小应约为 256 MB，分区应在 10 到 50 GB 之间。因此，最终配置取决于您正在使用的消息的数量和大小。在大多数情况下，您只需按摄取日期对数据进行分区就可以了。

<强>2。选中“不发出空文件选项”

您应该选中“不发出空文件选项”。如果想用Spark来消费数据，可以节省不必要的文件操作。

<强>3。使用文件路径中的数据源

使用流式架构，您的 EventHub 就相当于面向批处理的架构方法中的着陆区。因此，您将在原始数据层中摄取数据。良好的做法是在目录路径中使用数据源而不是 EventHub 的名称。例如，如果您从工厂的机器人获取遥测数据，这可能是目录路径 /raw/robots/

存储命名需要使用所有属性，例如 {Namesapce}、{PartitionId}。因此，最终，具有明确定义的路径、每日分区以及使用 Azure Data Lake Gen 2 中文件名的其余属性的良好捕获文件格式定义可能如下所示:

 /raw/robots/ingest_date={Year}-{Month}-{Day}/{Hour}{Minute}{Second}-{Namespace}-{EventHub}-{PartitionId}

<强>4。考虑压实工作

捕获的数据未压缩，并且在您的用例中也可能最终形成小文件(因为最小写入频率为 15 分钟)。因此，如果有必要，请编写每天运行一次的压缩作业。类似的东西

df.repartition(5).write.format("avro").save(targetpath)

将完成这项工作。

那么现在读取捕获数据的最佳实践是什么？

<强>5。忽略读取数据的非 avro 文件

Azure EventHubs Capture 将临时数据写入 Azure Data Lake Gen1。最佳实践是仅使用 avro 扩展名读取数据。您可以通过 Spark 配置轻松实现此目的:

spark.conf.set("avro.mapred.ignore.inputs.without.extension", "true")

<强>6。只读取相关分区

考虑仅读取相关分区，例如。 G。过滤当前摄取日。

<强>7。使用共享元数据

读取捕获的数据的工作方式与直接从 Azure EventHub 读取数据类似。所以你必须有一个模式。假设您还有使用 Spark 结构化流直接读取数据的作业，一个好的模式是存储元数据并共享它。您可以将此元数据存储在 Data Lake Store json 文件中:

[{"MeasurementTS":"timestamp","Location":"string", "Temperature":"double"}]

然后用这个 simple parsing function 来阅读它:

# parse the metadata to get the schema
from collections import OrderedDict 
from pyspark.sql.types import *
import json

ds = dbutils.fs.head (metadata)                                                 # read metadata file

items = (json
  .JSONDecoder(object_pairs_hook=OrderedDict)
  .decode(ds)[0].items())

#Schema mapping 
mapping = {"string": StringType, "integer": IntegerType, "double" : DoubleType, "timestamp" : TimestampType, "boolean" : BooleanType}

schema = StructType([
    StructField(k, mapping.get(v.lower())(), True) for (k, v) in items])

所以你可以重用你的架构:

from pyspark.sql.functions import *

parsedData = spark.read.format("avro").load(rawpath). \
  selectExpr("EnqueuedTimeUtc", "cast(Body as string) as json") \
 .select("EnqueuedTimeUtc", from_json("json", schema=Schema).alias("data")) \
 .select("EnqueuedTimeUtc", "data.*")