在 azure datafactory 管道中一一提供 azure databrick 活动中的文件名答案

【问题标题】：feeding filenames in azure databrick activity one by one in azure datafactory pipeline在 azure datafactory 管道中一一提供 azure databrick 活动中的文件名
【发布时间】：2019-06-17 11:27:50
【问题描述】：

我正在尝试创建一个数据工厂管道，其中一个活动将文件名（来自容器或其他文件夹）一一注入数据块活动中，以按传入顺序进行处理。如何实现？

【问题讨论】：

标签： azure-pipelines azure-data-factory-2 azure-databricks

【解决方案1】：

您好 Surbhi Tayal，感谢您的询问。按传入顺序，我假设您的意思是第一个发送到数据块的应该是第一个完成的，而不是并行处理。

要做到这一点，您将需要以下内容：

数组类型的管道变量。
一种用文件名填充数组变量的机制。它可能是 GetMetadata 活动或管道参数或其他东西。如果您需要这方面的帮助，请在问题中添加更多详细信息。
Databricks 资源和链接服务

在您的管道中创建一个 ForEach 活动。在设置中标记“顺序”选项。如果不这样做，将导致您的活动被并行发送，而不是一个接一个地发送。在设置中，引用“项目”中的数组变量。表达式看起来像@variables('myVariableName')。在 ForEach 活动的活动中，放置一个 Databricks 类型的活动。选项是“笔记本”、“Jar”和“Python”。对于我来说，我用的是笔记本。由于 UI 的“浏览”选项，Notebook 更易于设置。将活动设置为首先使用适当的链接服务。设置“Python 文件”/“笔记本路径”/“主类名”。展开“参数”部分并添加一个新参数。为参数指定与 Databricks 脚本中相同的名称。该值应为@string(item())（如果您的可枚举不是简单的原语数组，则可能会有所不同）。这会从 ForEach 活动中获取项目并确保它是一个字符串。必要时设置库。

当您尝试运行/调试时，请注意 Databricks 可能需要很长时间才能启动集群。这会增加管道运行时间。

【讨论】：

【解决方案2】：

根据源类型和文件被摄取的频率，它也可能是使用 spark 结构化流的一个选项。对于流数据源，还支持文件作为源 - 将写入目录中的文件作为数据流读取。支持的文件格式是文本、csv、json、orc、parquet（请参阅 DataStreamReader 接口的文档以获取最新列表以及每种文件格式支持的选项）。请注意，文件必须以原子方式放置在给定目录中，在大多数文件系统中，这可以通过文件移动操作来实现。

streamingInputDF = (
  spark
    .readStream           # Similar to Batch just using `readStream` instead of `read`
    .schema(jsonSchema)               
    .json(inputPath)
)

如果您不想永久运行笔记本，请使用触发一次选项。使用 trigger once 选项，输出为可用数据写入一次，没有此选项输出流将永久运行：

streamingOutputDF \
    .coalesce(1) \
    .writeStream \
    .format("parquet") \
    .partitionBy('ingest_date') \
    .option("checkpointLocation", checkPointPath) \
    .option("path", targetPath) \
    .trigger(once=True) \
    .start()

在这种情况下，您可以使用数据工厂来触发不带参数的 Databricks 笔记本。

【讨论】：