Boys，Grils，Friends，我是你的师哥or师弟，也可以是师姐or师妹哟！
领域展开：Java，Python，数据分析，数据可视化，大数据开发…
兴趣展开：阅读，旅游，运动，王者农药…

【今日重点：Azure的简介和使用步骤】

简介

Azure 是一个集存储和ETL的云平台，可以在此平台上可以进行数据的迁移、数据的etl以及数据的分析。
数据源支持一百多种，如常见的数据库（关系型、非关系型）、文件系统、hdfs、ftp文件、hive、hbase等；
数据的ETL支持hive，spark，pig，impala，mapreduce,以及已分装好的其他数据处理的功能；
数据的分析可以使用已分装好的功能(已经非常完善)，亦可以写自定义。

下面是如何使用azure 的factory来解决问题，大致提纲
先简单说一下步骤，感兴趣的、有条件的可以去实操一下：
创建数据工厂（DataFactory）
创建链接服务（源端和目标端的配置信息）
创建输入端和输出端数据源（dataset）
创建活动（具体的操作）
创建管道（运行活动的地方）
创建管道运行（管道实例）
创建监控管道运行（可选）
启动（或者叫部署）
可以在监控面板查看管道运行情况

1. 创建数据工厂的常用方式

使用用户界面（UI）
使用"复制数据"工具（UI）
- 1.新建容器
- 2.创建数据工厂
- 3.启动数据复制工具
使用python

2. 数据的接入、ETL、Analysis方式

2.1 移动/复制数据

使用数据复制工具（UI）
- 数据库全表复制
- 自定义筛选复制
- 可定时配置复制
使用Python SDK

2.3 转换数据

使用Databricks python活动
使用HDInsight Hive活动
使用HDInsight Spark活动

2.4 分析数据

-分组、统计、极值、去重
- 更改行：对行的插入、删除、更新和更新插入
- 有条件拆分：分流
- 派生列：添加列
- Exists：类似与sql exists
- 筛选：类似与sql where
- 平展：获取层次结构数组值并展开到行
- 连接：类似与 sql join
- 查找：类似于sql left join
- 新建分支：一源，多接收器使用
- 分析：分析数据中文档形式的列，如json、xml、规则文本
- 透视：类似与 sql行转列
- 级别：排序
- Select:对列进行重命名、删除、重新排序等
- 接收器：可定义多个
- 排序：可选列进行升序、降序排列
- 源转换：至少需要一个
- 代理键：类似于sql的自增主键（非业务键）
- Union:多个数据流组合成一个数据流
- 逆透视：类似于sql的列转行
- 窗口：类似于flink的时间或统计窗口，如LEAD、LAG、NTILE、CUMEDIST、RANK

3.python使用的完整示例

"""
使用python来使用DF，以Azure blob文件-> Azure blob文件的复制 为例
"""
from azure.identity import ClientSecretCredential 
from azure.mgmt.resource import ResourceManagementClient
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.mgmt.datafactory.models import *
from datetime import datetime, timedelta
import time

# 打印概览信息
def print_item(group):
    """Print an Azure object instance."""
    print("\tName: {}".format(group.name))
    print("\tId: {}".format(group.id))
    if hasattr(group, 'location'):
        print("\tLocation: {}".format(group.location))
    if hasattr(group, 'tags'):
        print("\tTags: {}".format(group.tags))
    if hasattr(group, 'properties'):
        print_properties(group.properties)
# 打印配置信息
def print_properties(props):
    """Print a ResourceGroup properties instance."""
    if props and hasattr(props, 'provisioning_state') and props.provisioning_state:
        print("\tProperties:")
        print("\t\tProvisioning State: {}".format(props.provisioning_state))
    print("\n\n")
# 打印活动详细信息（在运行）
def print_activity_run_details(activity_run):
    """Print activity run details."""
    print("\n\tActivity run details\n")
    print("\tActivity run status: {}".format(activity_run.status))
    if activity_run.status == 'Succeeded':
        print("\tNumber of bytes read: {}".format(activity_run.output['dataRead']))
        print("\tNumber of bytes written: {}".format(activity_run.output['dataWritten']))
        print("\tCopy duration: {}".format(activity_run.output['copyDuration']))
    else:
        print("\tErrors: {}".format(activity_run.error['message']))

# 启动方法
def main():
    # 0.定义必要参数
    subscription_id = '<subscription ID>'
    rg_name = '<resource group>'
    df_name = '<factory name>'

    credentials = ClientSecretCredential(client_id='<service principal ID>', client_secret='<service principal key>', tenant_id='<tenant ID>') 
    resource_client = ResourceManagementClient(credentials, subscription_id)
    adf_client = DataFactoryManagementClient(credentials, subscription_id)

    rg_params = {'location':'westus'}
    df_params = {'location':'westus'}
 
    # 1.创建资源组（不存在的情况）
    resource_client.resource_groups.create_or_update(rg_name, rg_params)

    # 2.创建数据工厂
    df_resource = Factory(location='westus')
    df = adf_client.factories.create_or_update(rg_name, df_name, df_resource)
    print_item(df)
    while df.provisioning_state != 'Succeeded':
        df = adf_client.factories.get(rg_name, df_name)
        time.sleep(1)

    # 3.创建存储链接服务
    ls_name = 'storageLinkedService001'

    storage_string = SecureString(value='DefaultEndpointsProtocol=https;AccountName=<account name>;AccountKey=<account key>;EndpointSuffix=<suffix>')
    ls_azure_storage = LinkedServiceResource(properties=AzureStorageLinkedService(connection_string=storage_string))
    
    ls = adf_client.linked_services.create_or_update(rg_name, df_name, ls_name, ls_azure_storage)
    print_item(ls)

    # 4.创建输入的dataset
    ds_name = 'ds_in'
    ds_ls = LinkedServiceReference(reference_name=ls_name)
    blob_path = '<container>/<folder path>'
    blob_filename = '<file name>'
    ds_azure_blob = DatasetResource(properties=AzureBlobDataset(
        linked_service_name=ds_ls, folder_path=blob_path, file_name=blob_filename))
    ds = adf_client.datasets.create_or_update(
        rg_name, df_name, ds_name, ds_azure_blob)
    print_item(ds)

    # 5.创建输出的dataset
    dsOut_name = 'ds_out'
    output_blobpath = '<container>/<folder path>'
    dsOut_azure_blob = DatasetResource(properties=AzureBlobDataset(linked_service_name=ds_ls, folder_path=output_blobpath))
    dsOut = adf_client.datasets.create_or_update(
        rg_name, df_name, dsOut_name, dsOut_azure_blob)
    print_item(dsOut)

    # 6.创建复制活动
    act_name = 'copyBlobtoBlob'
    blob_source = BlobSource()
    blob_sink = BlobSink()
    dsin_ref = DatasetReference(reference_name=ds_name)
    dsOut_ref = DatasetReference(reference_name=dsOut_name)
    copy_activity = CopyActivity(name=act_name, inputs=[dsin_ref], outputs=[
                                 dsOut_ref], source=blob_source, sink=blob_sink)

    # 7.创建管道
    p_name = 'copyPipeline'
    params_for_pipeline = {}
    p_obj = PipelineResource(
        activities=[copy_activity], parameters=params_for_pipeline)
    p = adf_client.pipelines.create_or_update(rg_name, df_name, p_name, p_obj)
    print_item(p)

    # 8.创建管道运行
    run_response = adf_client.pipelines.create_run(rg_name, df_name, p_name, parameters={})

    # 9.监控管道运行
    time.sleep(30)
    pipeline_run = adf_client.pipeline_runs.get(
        rg_name, df_name, run_response.run_id)
    print("\n\tPipeline run status: {}".format(pipeline_run.status))
    filter_params = RunFilterParameters(
        last_updated_after=datetime.now() - timedelta(1), last_updated_before=datetime.now() + timedelta(1))
    query_response = adf_client.activity_runs.query_by_pipeline_run(
        rg_name, df_name, pipeline_run.run_id, filter_params)
    print_activity_run_details(query_response.value[0])


# 启动
main()

原文链接：https://blog.csdn.net/qq_41200768/article/details/119899879

微软云计算平台Azure的学习教程

目录

简介