将 CSV 加载到 Redshift 表中，其中一些 CSV 缺少一些列

Question

目标

将 S3 存储桶中 CSV 文件中的所有数据加载到 Redshift 表中。

问题

某些文件缺少列的子集。

示例

在现实世界中，我的存储桶每天都会获得新的 CSV，但请考虑这个更简单的示例。

假设我有一个fruit 表：

id	name	val1	val2
INTEGER	VARCHAR	VARCHAR	INTEGER

假设我有 2 个 CSV。

test1.csv:

id	name	val1	val2
1	apple	abc	123
2	orange	def	456

test2.csv:

id	name	val2
3	banana	421
4	pear	68

请注意，test2.csv 缺少 val1 列。

问题

有没有办法使用内置的 Redshift 功能和 SQL 将任意多个 CSV 加载到 Redshift 表中，其中某些文件可能缺少（非主键）列的任何子集？其他一些 AWS 服务会完成这项工作吗？或者我现在必须转向 Python 吗？这是 psycopg2 和/或 SQLAlchemy 的工作吗？

我读到的

我会使用COPY，但文档似乎说它不支持我的方案。 column list 必须与源数据中列名的顺序相匹配，但如果 any 列可能丢失，则该顺序不是恒定的，因此我相信此功能无法帮助我。缺失的列并不总是在数据集的末尾，所以我相信COPY 的FILLRECORD 参数对我没有帮助。

creating a temporary or external table 的想法很有趣，但我不确定它是否支持我的方案。我不会遇到同样的问题吗？

更新：我找到的解决方案

使用 AWS 的最简单的解决方案是 AppFlow。我确认即使源文件中缺少目标表中定义的列，它也会从 CSV 加载数据。在 Redshift 中创建表时，我将默认值设置为 NULL。

如果想用 AWS Glue 和 Python 做一些更复杂的事情，那么我确认 Pandas 或 SQLAlchemy 将加载缺少列的 CSV。 AWS Data Wrangler 也应该这样做，但我还没有测试过。

Answer 1

以防万一您开始寻找基于 python 的流处理选项：

from convtools.contrib.tables import Table

def get_s3_file_descriptors():
    raise NotImplementedError


def get_superset_of_columns():
    # building a superset of columns, reading just first rows
    columns = {}
    for f in get_s3_file_descriptors():
        for column in Table.from_csv(f, header=True).columns:
            columns[column] = None
    return columns


def get_rows():
    columns = get_superset_of_columns()
    for f in get_s3_file_descriptors():
        table = Table.from_csv(f, header=True)
        current_columns = set(table.columns)
        for column in columns:
            if column not in current_columns:
                table.update(column=None)
        yield from table.take(*columns).into_iter_rows(tuple)


get_rows()