azure - 在 Azure 中将 Avro 转换为 Parquet

Question

我每天在 Azure BLOB 帐户中创建数百个 avro 文件。这些 avro 文件包含具有不同架构的 JSON 对象。每个文件大小范围在 100KB 到 1 MB 之间。我需要将它们转换为 parquet 格式并存储在同一个 Azure BLOB 帐户中。

我的限制:我需要使用 Azure native 服务。我不想使用 Azure Functions，因为即使是单个文件也可能需要花费大量时间来转换，而 Azure Functions 无法处理。因此，不要将 Azure Functions 作为解决方案的核心。

考虑到所描述的要求/限制，最佳方法是什么？

示例输入avro:

{
    "SequenceNumber": 123, 
    "Offset": "21212", 
    "EnqueuedTimeUtc": "2/18/2022 3:54:39 PM", 
    "SystemProperties": 
        {
            "x-opt-kafka-key": "211223131", 
            "x-opt-enqueued-time": 12321312321
        }, 
    "Properties": 
        {
            "element1": "value1", 
            "element2": "value2"}, 
    "Body":"{ 
                        \"version\":\"1.0\",
                        \"type\":\"some-type\",
                        \"data\":{
                                            \"id\":\"some-id\",
                                            \"color\":\"some-color\"
                                            \"address\":{
                                                                    \"line1\":\"some-line\",
                                                                    \"line-2\":\"some-line\"                                        },
}}

附注我已经在 AWS 中使用 Glue 完成了此操作。但是，ADF 不能很好地处理复杂的 avro 文件。

非常感谢您的帮助/建议。

Answer 1

我做了一个非常类似的用例，使用 Azure Synapse Analytics 将 avro 转换为平面关系表。关键组件如下:

• Azure Data Lake Gen 2 - avro 文件存储在数据湖中
• Synapse Pipelines - 管道使用“获取元数据”事件获取有关文件的信息，并使用并行运行的“For Each”事件来处理每个文件来获取 avro 文件
• Synapse Notebooks - 参数化的 Spark 笔记本，它打开原始 avro 文件并仅使用几行 Scala 将其扁平化为关系形式。这可以很容易地进行调整以将文件保存为 Parquet 。

Azure 数据工厂 (ADF) 和 Databricks 将提供类似的架构。

对于成本，您必须针对您的场景进行尝试，但是 Azure Pricing Calculator可以给你一个入门的想法。对于 Synapse Notebook，它还取决于您使用的 Spark 池的大小。这些尺寸可以是小型、中型、大型、特大号和特大号，成本也相应增加。我建议尝试对较少数量的文件进行比较，然后您发现哪一个更容易使用、更快、更便宜等

对于我的示例，我们有一个包含两列的 avro 文件:标题和正文。我们只对包含 json 并将其保存到数据库的 body 列感兴趣。我们的 json 并不是特别复杂，但我们确实处理许多不同数量的列。一个简化的例子:

参数单元格

val pSourceFilePath = "abfss://somefilesyst<a href="https://stackoverflow.com/cdn-cgi/l/email-protection" class="__cf_email__" data-cfemail="3e5b537e4d51535b6d4a514c5f595b7f5d5d514b504a105a584d105d514c5b104957505a51494d10505b4a" rel="noreferrer noopener nofollow">[email protected]</a>/raw/somepath/" 

称重传感器

这个看似简单的 spark.read.load 语句读取我们特定的文件格式，但您可能需要更改为 spark.read.format("avro") -这实际上取决于您的文件:

// Read the parquet file and assign to dataframe
val df = spark.read.
    load(s"${pSourceFilePath}")

展开单元格

这又是一个看似简单的代码，使用星号语法 (*) 将所有 json 扩展为关系列 - 这适用于我们的格式，但我不确定它是否适用于您的格式:

// Expand the json column to many columns
val df2 = df.select($"body.*", input_file_name.as("filename"))

然后，您可以将数据帧(现在采用更熟悉的柱状格式)写入 Parquet 。