Airflow PostgresToGCSOperator以历元格式导出时间戳列

Question

Im使用PostgresToGCSOperator [v1.10和v2.0导出数据。我知道那个操作符中有一个映射。我已经改变了这一点，并作为一个定制运营商。如果Postgres有时间戳，那么在BQ中使用DATETIME作为时间戳Z使用时间戳

       1114: 'DATETIME',
        1184: 'TIMESTAMP',
        1082: 'DATE',
        1083: 'TIME',
        1005: 'INTEGER',
        1007: 'INTEGER',
        1016: 'INTEGER',
        20: 'INTEGER',
        21: 'INTEGER',
        23: 'INTEGER',
        16: 'BOOLEAN',
        700: 'FLOAT',
        701: 'FLOAT',
        1700: 'FLOAT'

但是，在以CSV格式导出数据时，它不会考虑这一点，而是以EPOCH格式导出值。我怎么修理它

Answer 1

还有一个最佳选择：为地面军事系统操作员定制气流Postgres。因此，您可以使用CSV或JSON，并且没有任何限制

在原始操作器上更换：

type_map = {
        1114: 'DATETIME',
        1184: 'TIMESTAMP',
        1082: 'DATE',
        1083: 'TIME',
        1005: 'INTEGER',
        1007: 'INTEGER',
        1016: 'INTEGER',
        20: 'INTEGER',
        21: 'INTEGER',
        23: 'INTEGER',
        16: 'BOOLEAN',
        700: 'FLOAT',
        701: 'FLOAT',
        1700: 'FLOAT',
    }

以及转换：

        if isinstance(value, (datetime.datetime, datetime.date)):
            return str(value)
        if isinstance(value, datetime.time):
            return str(value)
        if isinstance(value, Decimal):
            return float(value)
        return value

Answer 2

我找到了一个棘手的解决办法。我们可以使用row_to_json函数将postgres本身的输出转换为JSON。然后将其导出到地面军事系统

但也有一些小问题

1. schema file将只有一列。所以你必须处理它
2. 它有row_to_json作为头
3. 加载时使用源格式为NEWLINE_DELIMITED_JSON，但在导出过程中使用CSV，分隔符为|，否则将添加不必要的引号
4. 用板条箱装一个schema file manually并使用它
5. 使用max bad records as 1，因为头名称将不会被解析
6. 加载时使用autodetect=False

样本：

sql='SELECT row_to_json(t) FROM (SELECT * FROM mytbl) t'

输出：

row_to_json
{'id': 1, 'c1': '2020-10-16', 'c2': '10:20:30', 'c3': '2020-10-16T10:20:30', 'c4': '2020-10-16T04:50:30+00:00'}