在 Azure 中编排动态时间序列管道-平芜编程栈

原文：towardsdatascience.com/orchestrating-a-dynamic-time-series-pipeline-with-azure-data-factory-and-databricks-810819608231?source=collection_archive---------9-----------------------#2024-05-31

探索如何使用 Azure Data Factory（ADF）和 Databricks 构建、触发和参数化一个时间序列数据管道，并附有逐步教程。

https://medium.com/@johnleungTJ?source=post_page---byline--810819608231--------------------------------https://towardsdatascience.com/?source=post_page---byline--810819608231-------------------------------- John Leung

·发布于Towards Data Science ·阅读时间 8 分钟·2024 年 5 月 31 日

–

在上一篇故事中，我们回顾了 PySpark 在 Databricks 上处理时间序列数据的潜力。我鼓励你通过这里了解更多内容。在不配置独立 Spark 实例的情况下，我们可以通过 Databricks 上的 PySpark 摄取静态和流数据，执行数据转换，提取有用的时间相关特征，并构建可视化。当处理企业级数据的大规模复杂转换时，PySpark 的可扩展性和性能特别具有优势，甚至可以处理 PB 级别的数据。

所有特征工程任务都成功地在一个 Databricks 笔记本中完成。然而，这只是构建数据中心系统时数据工程故事的一部分。数据管道的核心部分在于数据编排。

数据编排通常指的是对数据流进行集中控制，以便我们可以自动化、管理和监控整个数据管道。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/2a39ebad8db26cfd20e2a647df768903.png

图片由Julio Rionaldo提供，来自Unsplash

# Define the configuration specificationsconfigs={"fs.azure.account.auth.type":"OAuth","fs.azure.account.oauth.provider.type":"org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider","fs.azure.account.oauth2.client.id":"<Client ID>","fs.azure.account.oauth2.client.secret":"<Client Secret>","fs.azure.account.oauth2.client.endpoint":"https://login.microsoftonline.com/<Tenant ID>/oauth2/token"}dbutils.fs.mount(source="abfss://input@adlstsdp.dfs.core.windows.net/",# URI of the object storagemount_point="/mnt/adlstsdp/input",# local path in the /mnt directoryextra_configs=configs)

为了验证文件访问，我们可以运行以下命令：

dbutils.fs.ls(“/mnt/adlstsdp/input”)# Output: [FileInfo(path='dbfs:/mnt/adlstsdp/input/household_power_consumption.csv', name='household_power_consumption.csv', size=132960755, modificationTime=1716798010000)]

#2 在 Azure Databricks 中嵌入 Notebook

本节中的大部分源代码基于我的上一篇文章。其思路是进行数据清理、转换和特征工程（创建时间相关特征和移动平均特征）。转换后的数据最终写入 Azure 数据库表中。

你可以查看下面的完整代码，了解其实现过程。

# Define file location, file typem and CSV optionsfile_location="/mnt/adlstsdp/input/household_power_consumption.csv"file_type="csv"schema="Date STRING, Time STRING, Global_active_power DOUBLE, Global_reactive_power DOUBLE, Voltage DOUBLE, Global_intensity DOUBLE, Sub_metering_1 DOUBLE, Sub_metering_2 DOUBLE, Sub_metering_3 DOUBLE"first_row_is_header="true"delimiter=";"# Read CSV filesorg_df=spark.read.format(file_type)\.schema(schema)\.option("header",first_row_is_header)\.option("delimiter",delimiter)\.load(file_location)# Data cleansing and transformationfrompyspark.sql.functionsimport*cleaned_df=org_df.na.drop()cleaned_df=cleaned_df.withColumn("Date",to_date(col("Date"),"d/M/y"))cleaned_df=cleaned_df.withColumn("Date",cleaned_df["Date"].cast("date"))cleaned_df=cleaned_df.select(concat_ws(" ",to_date(col("Date"),"d/M/y"),col("Time")).alias("DateTime"),"*")cleaned_df=cleaned_df.withColumn("DateTime",cleaned_df["DateTime"].cast("timestamp"))df=cleaned_df.groupby("Date").agg(round(sum("Global_active_power"),2).alias("Total_global_active_power"),).sort(["Date"])# Add time-related featuresdf=df.withColumn("year",year("Date"))df=df.withColumn("month",month("Date"))df=df.withColumn("week_num",weekofyear("Date"))# Add lagged value features of total global active powerfrompyspark.sql.windowimportWindowfrompyspark.sql.functionsimportlag windowSpec=Window.orderBy("Date")df=df.withColumn("power_lag1",round(lag(col("Total_global_active_power"),1).over(windowSpec),2))# Create delta fielddf=df.withColumn("power_lag1_delta",round(col("power_lag1")-col("Total_global_active_power"),2))# Create window average fieldsdefadd_window_avg_fields(df,window_sizes):foridx,window_sizeinenumerate(window_sizes,start=1):window_col_name=f"avg_power_lag_{idx}"windowSpec=Window.orderBy("Date").rowsBetween(-window_size,0)df=df.withColumn(window_col_name,round(avg(col("Total_global_active_power")).over(windowSpec),2))returndf window_sizes=[14,30]df=add_window_avg_fields(df,window_sizes)# Create Exponentially Weighted Moving Average (EWMA) fieldsimportpyspark.pandasasps ps.set_option('compute.ops_on_diff_frames',True)defadd_ewma_fields(df,alphas):foridx,alphainenumerate(alphas,start=1):ewma_col_name=f"ewma_power_weight_{idx}"windowSpec=Window.orderBy("Date")df[ewma_col_name]=df.Total_global_active_power.ewm(alpha=alpha).mean().round(2)returndf alphas=[0.2,0.8]df_pd=df.pandas_api()df_pd=add_ewma_fields(df_pd,alphas)df=df_pd.to_spark()# Write transformed dataframe to the database table "electric_usage_table"df.write.format("jdbc")\.option("url","jdbc:sqlserver://sql-db-dp.database.windows.net:1433;databaseName=sql-db-dp")\.option("dbtable","dbo.electric_usage_table")\.option("user","<username>")\.option("password","<password>")\.mode("overwrite")\.save()

#3 在 ADF 中构建基本管道

在 ADF 中，我们将“Notebook”活动添加到管道环境中，然后配置它以引用 Databricks 文件夹中的所需 Notebook。设置 Databricks 连接服务，然后在 ADF 中验证并发布整个活动管道。然后，您可以在“调试”模式下运行管道。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/77f80f4bd8dac29b014997ad560cb4d7.png

管道运行的成功状态（图片由作者提供）

活动状态显示为“已成功”，这意味着数据应该已迁移并插入到 Azure SQL 数据库表中。我们可以使用查询编辑器查看结果以进行验证。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/f9a3cbd5c05477cfd6c63700ff13670f.png

查询 Azure SQL 数据库的结果（图片由作者提供）

#4 自动化管道

ADF 提供的功能远超上述简单实现。例如，我们可以通过创建基于存储的事件触发器来自动化管道。确保Microsoft.EventGrid已注册为您账户订阅中的资源提供者之一，然后设置触发器：每当新数据集上传到存储帐户时，管道将自动执行。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/cd7dbcf9ffdcf90150c136a408b6bf58.png

在 ADF 中设置新的触发器（图片由作者提供）

这种类型的触发器在行业中有各种应用场景，例如监控库存水平以补充供应链订单，或追踪客户互动以实现数字营销中的个性化推荐。

#5 参数化 Notebook 变量

为了进一步构建更具动态性的数据信息管道，我们可以使变量更加参数化。例如，在时间序列数据的特征工程中，数据特征的窗口大小最初可能并未优化。窗口大小可能需要根据季节性模式或下游模型微调进行调整。对于这种情况，我们可以通过以下设置进行修改。

https://github.com/OpenDocCN/towardsdatascience-blog-zh-2024/raw/master/docs/img/3c3140b421b91d991eb48967cfdc610b.png

设置管道运行的参数（图片由作者提供）

在 Notebook 中，添加以下代码以创建一个小部件，可以从 ADF 管道获取参数输入：

# Additional code: Access the current value of the widgetinputWindowSizes=dbutils.widgets.get("inputWindowSizes")window_sizes=inputWindowSizes.split(",")# Original function for adding window average featuresdf=add_window_avg_fields(df,window_sizes)