使用 Apache Beam(Direct Runner)将 TIMESTAMP 写入 PostgreSQL 的正确方法是什么?我在任何地方都找不到这个记录。我尝试将日期格式化为rfc3339
字符串,如下所示并使用 Python SDK 编写apache_beam.io.jdbc.WriteToJdbc
无济于事。我的管道失败并出现以下错误:
Caused by: java.sql.BatchUpdateException: Batch entry 0 INSERT INTO beam_direct_load VALUES('Product_0993', 'Whse_J', 'Category_028', '2012-07-27T00:00:00', 100) was aborted: ERROR: column "date" is of type timestamp without time zone but expression is of type character varying
该表定义如下:
CREATE TABLE IF NOT EXISTS public.beam_direct_load(
product_code VARCHAR(255),
warehouse VARCHAR(255),
product_category VARCHAR(255),
date TIMESTAMP,
order_demand INTEGER
);
我已经为此注册了编码器ProductDemand
:
class ProductDemand(typing.NamedTuple):
product_code: str
warehouse: str
product_category: str
date: str
order_demand: int
coders.registry.register_coder(ProductDemand, coders.RowCoder)
我的管道定义如下:
(
pipeline
| 'ExtractFromText' >> ReadFromText(input_file, skip_header_lines=1)
| 'Split' >> Map(lambda x: [element.strip() for element in x.split(',')])
| 'DropNA' >> Filter(lambda x: x[3] != 'NA' )
| 'FormatData' >> Map(lambda x:
[
x[0],
x[1],
x[2],
datetime.strftime(datetime.strptime(x[3], '%Y/%m/%d'), '%Y-%m-%dT%H:%M:%S'),
int(x[4].replace('(', '').replace(')', ''))
]
)
| 'MapToDBRow' >> Map(lambda x: ProductDemand(product_code=x[0], warehouse=x[1], product_category=x[2], date=x[3], order_demand=x[4])).with_output_types(ProductDemand)
| 'LoadToPostgres' >> WriteToJdbc
(
table_name='beam_direct_load',
driver_class_name='org.postgresql.Driver',
jdbc_url='jdbc:postgresql://localhost:5432/{}'.format(pg_db),
username=pg_username,
password=pg_password,
)
)