在Azure Synapse Pipeline中生成2018至当前年度季度日期数组

需求说明 #

需要生成2018年至当前日期的所有季度末日期数组（格式如yyyy-MM-dd），用于REST API批量拉取报表到ADLS，当前需覆盖至2024-03-31，避免手动维护数组。

解决方案一：直接用Synapse管道表达式生成（无需Notebook） #

利用Synapse管道的内置函数动态生成数组，无需依赖Spark计算，步骤如下：

1. 在管道中创建一个数组变量（比如quarterlyDates）
2. 将变量的默认值设置为以下表达式：

@filter(
    split(
        join(
            range(2018, int(formatDateTime(utcNow(), 'yyyy')) + 1),
            ',',
            concat(item(), '-03-31,', item(), '-06-30,', item(), '-09-30,', item(), '-12-31')
        ),
        ','
    ),
    lessOrEquals(item(), formatDateTime(utcNow(), 'yyyy-MM-dd'))
)

表达式说明： #

• range(2018, int(formatDateTime(utcNow(), 'yyyy')) +1)：生成2018到当前年份的整数序列
• join(...)：将每个年份的4个季度末日期拼接成逗号分隔的字符串
• split(...)：将字符串拆分为日期数组
• filter(...)：过滤掉晚于当前日期的季度末（比如当前是2024年5月，会自动排除2024-06-30及以后的日期）

解决方案二：改进PySpark + Lookup活动方案 #

如果你偏好使用Spark生成日期，可调整代码并配合Lookup活动将结果转为管道数组：

步骤1：修改PySpark代码（高效生成季度末） #

替换原遍历每天的低效逻辑，直接生成季度末并过滤，最后输出为结构化JSON：

from pyspark.sql.functions import current_date
import json

def generate_quarterly_dates(start_year=2018):
    current_dt = current_date().strftime("%Y-%m-%d")
    current_year = int(current_dt[:4])
    
    # 直接生成所有季度末日期
    quarterly_dates = []
    for year in range(start_year, current_year + 1):
        quarterly_dates.extend([
            f"{year}-03-31",
            f"{year}-06-30",
            f"{year}-09-30",
            f"{year}-12-31"
        ])
    
    # 过滤超过当前日期的记录
    filtered_dates = [date for date in quarterly_dates if date <= current_dt]
    
    # 写入ADLS为单条JSON（包含dates数组）
    output_path = "/dbfs/mnt/your-adls-container/path/quarterly_dates.json"
    with open(output_path, "w") as f:
        json.dump({"dates": filtered_dates}, f)

generate_quarterly_dates()

步骤2：用Lookup活动读取并转为管道数组 #

1. 在管道中添加Lookup活动，配置读取上述JSON文件
2. 在管道变量赋值时，使用以下表达式提取数组：

@activity('LookupQuarterlyDates').output.firstRow.dates

针对原问题的修复（如果保留原Spark输出格式） #

如果仍使用原Spark代码生成每行一个日期的JSON，Lookup活动返回的output.value是包含每行对象的数组，可通过map函数提取日期字符串：

@map(activity('LookupOldDates').output.value, item().date_string)

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