Load CSVs into Redshift table, where some CSVs lack some columns

Question

Goal

Load all data from CSV files in an S3 bucket into a Redshift table.

Problem

Some files lack a subset of columns.

Example

In the real world, my bucket gets new CSVs daily, but consider this simpler example.

Suppose I have a fruit table:

id	name	val1	val2
INTEGER	VARCHAR	VARCHAR	INTEGER

Suppose I have 2 CSVs.

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

Note that test2.csv lacks the val1 column.

Question

Is there any way to load arbitrarily many CSVs into a Redshift table, where some files may lack any subset of (non-primary key) columns, using built-in Redshift functionality and SQL? Will some other AWS service(s) get the job done? Or must I turn to Python at this point? Is this a job for psycopg2 and/or SQLAlchemy?

What I've read

I'd use COPY, but the docs seem to say it doesn't support my scenario. A column list must match the order of column names in the source data, but if any columns can be missing, then that order isn't constant, so I believe this functionality can't help me. The missing columns aren't always at the end of the dataset, so I believe COPY's FILLRECORD parameter won't help me.

The idea of creating a temporary or external table is interesting, but I'm unsure if it supports my scenario. Wouldn't I just run into the same issues?

Update: Solutions I found

The simplest solution using AWS is AppFlow. I confirmed that it will load data from CSV even if columns defined in the destination table are missing from a source file. I set the default values to NULL when creating the table in Redshift.

If one wants to do something more complicated with AWS Glue and Python, then I confirmed Pandas or SQLAlchemy will load CSVs with missing columns. AWS Data Wrangler should also do it, but I haven't tested that.

user_5387738 · Answer

Just in case you start looking for python-based stream processing options:

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()