如何通过AWS Data Exchange API或Python获取AWS Marketplace免费开放数据集

问题分析 #

你调用list_data_sets接口返回空结果，核心原因是这个接口默认仅返回当前AWS账户已订阅的数据集。要获取AWS Marketplace中无需订阅的公开开放数据集，需要调整API调用参数并筛选符合条件的条目。

解决方案：Python调用AWS Data Exchange API实现 #

通过指定Origin参数为AWS_MARKETPLACE，并筛选SubscriptionRequired=False的数据集，同时处理API的分页返回（单次调用最多返回100条），即可获取目标数据集及其详情。

修正后的Python代码如下：

import boto3

# 初始化Data Exchange客户端（建议通过AWS凭证文件/环境变量/IAM角色管理密钥，避免硬编码）
client = boto3.client('dataexchange', region_name='us-east-1')

def list_public_open_datasets():
    public_datasets = []
    next_token = None
    
    # 处理分页，遍历所有结果
    while True:
        request_params = {'Origin': 'AWS_MARKETPLACE'}
        if next_token:
            request_params['NextToken'] = next_token
        
        response = client.list_data_sets(**request_params)
        
        # 筛选无需订阅的数据集
        for dataset in response['DataSets']:
            if not dataset.get('SubscriptionRequired', True):
                public_datasets.append(dataset)
        
        next_token = response.get('NextToken')
        if not next_token:
            break
    
    return public_datasets

# 获取并打印数据集核心信息
if __name__ == '__main__':
    datasets = list_public_open_datasets()
    print(f"找到{len(datasets)}个无需订阅的开放数据集：")
    for idx, ds in enumerate(datasets, 1):
        print(f"\n--- 数据集{idx} ---")
        print(f"名称: {ds['Name']}")
        print(f"ID: {ds['Id']}")
        print(f"描述: {ds.get('Description', '无描述')}")
        print(f"数据类型: {ds['AssetType']}")

关键说明 #

• Origin='AWS_MARKETPLACE'参数：限定只返回AWS Marketplace中的数据集，排除自定义或其他来源的数据集。
• SubscriptionRequired=False筛选：精准过滤出无需订阅即可访问的开放数据集。
• 分页处理：list_data_sets接口单次最多返回100条数据，通过NextToken循环迭代获取全量结果。

补充：获取数据集的具体资产信息 #

如果需要查看数据集内的具体资源（如文件、API端点等），可调用list_data_set_assets接口，传入目标数据集的Id：

def get_dataset_assets(dataset_id):
    assets = []
    next_token = None
    while True:
        request_params = {'DataSetId': dataset_id}
        if next_token:
            request_params['NextToken'] = next_token
        
        response = client.list_data_set_assets(**request_params)
        assets.extend(response['Assets'])
        
        next_token = response.get('NextToken')
        if not next_token:
            break
    return assets

# 示例：获取第一个数据集的资产列表
if datasets:
    first_ds_id = datasets[0]['Id']
    assets = get_dataset_assets(first_ds_id)
    print(f"\n{datasets[0]['Name']}的资产列表：")
    for asset in assets:
        print(f"- 名称: {asset['Name']}, 类型: {asset['AssetType']}")

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