Один шаблон, реализующий путь холодного проглатывания, использует Event Hubs Capture . При захвате событий EventHubs записывает по одному файлу на раздел, как определено в параметрах . Данные записаны в формате avro и могут быть проанализированы с помощью Apache Spark.
Итак, каковы лучшие практики использования этой функции?
1. Не преувеличивайте
Часто я видел людей, использующих конфигурацию по умолчанию, что в итоге часто приводит к множеству маленьких файлов. Если вы хотите использовать данные, загруженные через EventHubs Capture с Spark, имейте в виду рекомендации для размеров файлов в хранилище озера данных Azure и разделов с Spark. Размер файла должен составлять ~ 256 МБ, а разделов - от 10 до 50 ГБ. Итак, наконец, конфигурация зависит от количества и размеров сообщений, которые вы потребляете. В большинстве случаев у вас все в порядке, просто разбивая данные по дате загрузки.
2. Установите флажок «Не выбрасывать пустые файлы»
Вы должны установить флажок «Не выбрасывать пустые файлы». Если вы хотите использовать данные с Spark, который сохраняет ненужные файловые операции.
3. Используйте источник данных в своих файловых путях
С потоковой архитектурой ваш EventHub - это то, что Landing Zone будет в подходе пакетно-ориентированной архитектуры. Таким образом, вы будете получать данные в слое raw-data. Хорошей практикой является использование источников данных вместо имени 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.conf.set("avro.mapred.ignore.inputs.without.extension", "true")
6. Только для чтения соответствующих разделов
Рекомендуется читать только соответствующие разделы, например, фильтровать текущий день приема.
7. Использовать общие метаданные
Чтение захваченных данных работает аналогично чтению данных непосредственно из Azure EventHubs. Таким образом, вы должны иметь схему. Предполагая, что у вас также есть задания по чтению данных непосредственно с помощью Spark Structured Streaming, хорошим шаблоном является сохранение метаданных и их совместное использование. Вы можете просто сохранить эти метаданные в json-файле Data Lake Store:
[{"MeasurementTS":"timestamp","Location":"string", "Temperature":"double"}]
и прочитать его с помощью этой простой функции синтаксического анализа :
# 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.*")