如何通过Notebook将Azure Data Lake数据导入Microsoft Fabric Lakehouse？

场景说明 #

Azure存储中存储着来自Dynamics 365 F&O的表数据，包含记录列名和类型的JSON头文件（.cdm.json），以及对应的单张或多张CSV数据文件。需要将每个表的头文件与数据文件结合后，导入至Microsoft Fabric Lakehouse。

当前尝试代码 #

import json
import os

def get_cdm_files(directory_path):
    cdm_files = []

    for root, diers, files in os.walk(directory_path):
        for file in files:
            if file.endswith('.cdm.json'):
                cdm_files.append(os.path.join(root, file))
  
    return cdm_files

def load_table_cdm_file(cdm_file_path):
    with open(cdm_file_path.replace("abfss://rootfoldername@datalakename.dfs.core.windows.net/", "/dbfs/mnt/dynamics/")) as f:
        cdm_json = json.load(f)

    colss = []
    for item in cdm_json['definitions'][0]['hasAttributes']:
        colss.append(item["name"])
  
    return spark.read.csv(cdm_file_path.replace("cdm.json", ".csv"), header=False, inferSchema=True)

def load_all_tables(cdm_files):
    tables = {}

    for cdm_file in cdm_files:
        table_name = cdm_file.split("/")[-1].replace(".cdm.json", "").lower()
        tables[table_name] = load_table_cdm_file(cdm_file)
    
    return tables

def write_table_delta(table_name, table_df):
    spark.sql(f"DROP TABLE IF EXISTS Lakehousename.Dynamics365_{table_name}")
    table_df.write.mode("overwrite").format("delta").saveAsTable(f"Dynamics365_{table_name}")

def main():
    cdm_files = get_cdm_files("abfss://rootfoldername@datalakename.dfs.core.windows.net/domainname.operations.dynamics.com/Tables/")
    
    if "TABLENAME1.cdm.json" in cdm_files:
        cdm_files.remove("abfss://rootfoldername@datalakename.dfs.core.windows.net/domainname.operations.dynamics.com/Tables/Custom/TABLENAME1.cdm.json")
    
    if "TABLENAME2.cdm.json" in cdm_files:
        cdm_files.remove("abfss://rootfoldername@datalakename.dfs.core.windows.net/domainname.operations.dynamics.com/Tables/Custom/TABLENAME2.cdm.json")
    
    if "TABLE3.cdm.json" in cdm_files:
        cdm_files.remove("abfss://rootfoldername@datalakename.dfs.core.windows.net/domainname.operations.dynamics.com/Tables/Custom/TABLE3.cdm.json")

    tables = load_all_tables(cdm_files)

    for table_name, table_df in tables.items():
        write_table_delta(table_name, table_df)

问题 #

尝试查找相关指南，但因该场景较新，可用资源极少，AI也未能提供有效帮助，恳请协助解决此数据导入问题。

修正与优化方案 #

原始代码存在ABFSS路径兼容、Schema未应用、多CSV文件处理等问题，以下是修正后的完整实现：

修正后的代码 #

import json
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, DateType, TimestampType

def get_cdm_files(directory_path):
    # 使用Spark遍历ABFSS路径下的CDM文件
    file_list = spark.sql(f"SELECT path FROM delta.`{directory_path}` WHERE path LIKE '%.cdm.json'").collect()
    return [row.path for row in file_list]

def map_cdm_type_to_spark(cdm_type):
    # 映射CDM数据类型到Spark数据类型
    type_mapping = {
        "string": StringType(),
        "integer": IntegerType(),
        "double": DoubleType(),
        "date": DateType(),
        "datetime": TimestampType(),
        "boolean": StringType()  # 若CDM中存在布尔型，可根据实际情况调整为BooleanType()
    }
    return type_mapping.get(cdm_type.lower(), StringType())

def load_table_cdm_file(cdm_file_path):
    # 读取CDM JSON文件（使用Spark读取ABFSS路径）
    cdm_json_str = spark.read.text(cdm_file_path, wholetext=True).first()[0]
    cdm_json = json.loads(cdm_json_str)
    
    # 基于CDM定义构建Spark Schema
    schema_fields = []
    for attr in cdm_json['definitions'][0]['hasAttributes']:
        col_name = attr["name"]
        cdm_type = attr.get("dataType", "string")
        spark_type = map_cdm_type_to_spark(cdm_type)
        schema_fields.append(StructField(col_name, spark_type, nullable=True))
    
    schema = StructType(schema_fields)
    
    # 读取对应CSV文件，支持单文件和多文件场景
    csv_path = cdm_file_path.replace(".cdm.json", ".csv")
    try:
        df = spark.read.csv(csv_path, header=False, schema=schema, sep=",", quote='"', escape='"')
    except Exception:
        # 若单文件不存在，读取CDM所在目录下的所有CSV
        csv_dir = os.path.dirname(cdm_file_path)
        df = spark.read.csv(csv_dir, header=False, schema=schema, sep=",", quote='"', escape='"')
    
    return df

def load_all_tables(cdm_files, exclude_tables):
    tables = {}
    for cdm_file in cdm_files:
        table_name = os.path.basename(cdm_file).replace(".cdm.json", "").lower()
        if table_name in exclude_tables:
            continue
        tables[table_name] = load_table_cdm_file(cdm_file)
    return tables

def write_table_delta(table_name, table_df, lakehouse_schema):
    full_table_name = f"{lakehouse_schema}.Dynamics365_{table_name}"
    spark.sql(f"DROP TABLE IF EXISTS {full_table_name}")
    table_df.write.mode("overwrite").format("delta").saveAsTable(full_table_name)

def main():
    # 配置参数
    storage_path = "abfss://rootfoldername@datalakename.dfs.core.windows.net/domainname.operations.dynamics.com/Tables/"
    lakehouse_schema = "Lakehousename"
    exclude_tables = {"tablename1", "tablename2", "table3"}  # 统一小写，避免大小写问题
    
    # 获取所有CDM文件
    cdm_files = get_cdm_files(storage_path)
    
    # 加载所有需要的表
    tables = load_all_tables(cdm_files, exclude_tables)
    
    # 写入Lakehouse
    for table_name, table_df in tables.items():
        write_table_delta(table_name, table_df, lakehouse_schema)

if __name__ == "__main__":
    main()

核心改进点 #

1. ABFSS路径兼容：改用Spark SQL查询文件路径，替代仅支持本地文件系统的os.walk，直接适配Azure存储的ABFSS协议。
2. 精准Schema构建：基于CDM文件中的数据类型定义映射Spark Schema，避免自动推断导致的类型错误，保证数据一致性。
3. 灵活CSV读取：先尝试读取单个CSV文件，失败则读取CDM所在目录下的所有CSV，适配单文件和分文件存储的场景。
4. 简化排除逻辑：将需要排除的表名放入集合，通过表名直接过滤，避免硬编码完整路径的冗余。
5. 参数化配置：集中管理存储路径、Lakehouse Schema、排除表等配置，提升代码可维护性。

内容的提问来源于stack exchange，提问作者Zephyr Harrison