Azure 数据工厂：从复制活动中输出复制的文件和文件夹信息答案

【问题标题】：Azure Data Factory: Output Copied File and Folder Information from Copy ActivityAzure 数据工厂：从复制活动中输出复制的文件和文件夹信息
【发布时间】：2019-05-27 20:46:49
【问题描述】：

我正在使用 Azure 数据工厂中的自托管集成运行时将数据从本地源（普通文件系统）复制到 Azure Blob 存储目标。传输后，我想通过附加在 Databricks 集群上运行的笔记本来自动处理文件。管道工作正常，但我的问题涉及复制活动的输出。

有没有办法获取每次运行时传输的文件和文件夹的信息？我会将这些信息作为参数传递给笔记本。

查看文档，似乎只有汇总信息可用：

https://docs.microsoft.com/en-us/azure/data-factory/copy-activity-overview

如果您传输大量文件，哪种方法更有意义。如果不可能，我想另一种方法是将复制过程留给自己，并根据存储帐户事件创建另一个管道？或者也许将每次运行的新文件和文件夹信息存储在一个固定的文本文件中，也可以传输它，然后在笔记本中读取？

【问题讨论】：

标签： azure copy azure-data-factory-2

【解决方案1】：

如果您想获取从数据工厂读取的文件或目录的信息，可以使用 Get Metadata Activity 来完成，请参阅以下answer 示例。

在笔记本中检测新文件的另一种方法是将结构化流与文件源结合使用。这很好用，您只需在复制活动之后调用笔记本活动。

为此，您定义一个流式输入数据帧：

streamingInputDF = (
   spark
     .readStream                     
     .schema(pqtSchema)               
     .parquet(inputPath) 
 )

inputPath 指向 Blob 存储中的输入目录。支持的文件格式是文本、csv、json、orc、parquet，因此这取决于您的具体情况。

重要的是在目标上您使用触发一次选项，因此笔记本不需要永久运行，例如。 g.：

streamingOutputDF \
    .repartition(1) \
    .writeStream \
    .format("parquet") \
    .partitionBy('Id') \
    .option("checkpointLocation", adlpath +  "spark/checkpointlocation/data/trusted/sensorreadingsdelta") \
    .option("path", targetPath + "delta") \
    .trigger(once=True) \
    .start()

另一种方法是使用 Azure 队列存储 (AQS)，请参阅以下documentation。

【讨论】：

我喜欢使用结构化流的想法。我将 CSV 文件作为输入，但它们的架构各不相同，并且分布在子文件夹中。我想我都明白了（推断模式、全局模式），但我很难将每个 CSV 写入相应的表（Hive）。我在 writeStream 上使用 foreachBatch，但它似乎将各种不相关的文件提取到同一个目标表中，而不仅仅是通过数据工厂更新的单个文件.我将对此进行进一步研究，并可能会在帖子中编辑有关此特定方法的详细信息。
好的，您的场景似乎更复杂。是的，将选择子目录中的所有文件，最初是所有文件。 php。您可以将该方法与数据工厂中的文件夹作为参数（用于输入/输出和检查点目录）混合使用吗？

【解决方案2】：

在这种情况下，解决方案实际上非常简单。我刚刚在 Azure 数据工厂中创建了另一个管道，它由 Blob Created 事件触发，folder 和 filename 作为参数传递给我的笔记本。似乎运行良好，并且需要最少的配置或代码。基本的过滤可以通过事件来完成，剩下的就看notebook了。

对于遇到这种情况的其他人，请参阅以下详细信息：

https://docs.microsoft.com/en-us/azure/data-factory/how-to-create-event-trigger

【讨论】：