在 Amazon Redshift 中,我有一个表,需要从多个 CSV 文件加载数据:
create table my_table (
id integer,
name varchar(50) NULL
email varchar(50) NULL,
processed_file varchar(256) NULL
);
前三列引用文件中的数据。最后一列
processed_filed
表示记录是从哪个文件导入的。
我在 Amazon S3 中有这些文件,我想使用
COPY
命令导入它们。比如:
COPY {table_name} FROM 's3://file-key'
WITH CREDENTIALS 'aws_access_key_id=xxxx;aws_secret_access_key=xxxxx'
DATEFORMAT 'auto' TIMEFORMAT 'auto' MAXERROR 0 ACCEPTINVCHARS '*' DELIMITER '\t' GZIP;
有没有办法使用 COPY 命令自动填充第四个
processed_file
列,以插入文件名。
我可以在 COPY 之后执行 UPDATE 语句,但我正在处理大量数据,因此理想情况下我希望尽可能避免这种情况。
在这里您可以尝试使用此自定义逻辑来添加新列,在此示例中将文件名添加为 redshift COPY 中的新列
import boto3
import re
s3 = boto3.client('s3')
sql = "DROPSQL , CREATE SQL , COPY SQL" ## here need to pass your actual sqls
def Filter(datalist,keyword):
# Search data based on regular expression in the list
return [val for val in datalist
if re.search(keyword, val)]
def add_new_col(table_name):
drop_sql = ''.join(Filter(sql.split(';'),keyword=table_name+' '))
create_sql = ''.join(Filter(sql.split(';'),keyword=table_name+'\('))
copy_sql = ''.join(Filter(sql.split(';'),keyword=table_name.upper()+'/'))
BUCKET = copy_sql.split(' ')[3].split('/')[2]
folder = '/'.join(copy_sql.split(' ')[3].split('/')[3:-1])+'/'
maintable = copy_sql.split(' ')[1]
print ("BUCKET {}, key_folder {}, maintable {}".format(BUCKET,folder,maintable))
temp_table_drop_sql = drop_sql.replace(table_name,'temp_table')
temp_table_create_sql = create_sql.replace(table_name,'temp_table')
temp_table_copy_sql = copy_sql.replace(table_name.upper(),'temp_table')
temp_table_name_withSchema = temp_table_copy_sql.split(' ')[1]
print ("temp_table_name_withSchema {}".format(temp_table_name_withSchema))
## replace with query execute logic
print(temp_table_drop_sql)
print(temp_table_create_sql)
#####
response = s3.list_objects_v2(
Bucket=BUCKET,
Prefix =folder)
new_column_name = 'filename'
for i in response["Contents"]:
## replace with query execute logic
temp_sql = copy_sql.replace(folder,i["Key"])
temp_sql = temp_sql.replace(table_name.upper(),'temp_table')
print(temp_sql)
## i["Key"] is filename
print("alter table {} ADD COLUMN {} varchar(256) NOT NULL DEFAULT '{}';".format(temp_table_name_withSchema, new_column_name , i["Key"].split('/')[-1]))
print("insert into {} (select * from {})".format(maintable, temp_table_name_withSchema))
print("truncate {}".format(temp_table_name_withSchema))
#####
## replace with query execute logic
print(drop_sql)
########
add_new_col(table_name)
其实是有可能的。我正在创建和加载数据,而无需额外的
processed_file_name
列,然后添加具有默认值的列。完整过程如下:
create table my_table (
id integer,
name varchar(50) NULL
email varchar(50) NULL,
);
COPY {table_name} FROM 's3://file-key'
WITH CREDENTIALS 'aws_access_key_id=xxxx;aws_secret_access_key=xxxxx'
DATEFORMAT 'auto' TIMEFORMAT 'auto' MAXERROR 0 ACCEPTINVCHARS '*' DELIMITER '\t' GZIP;
ALTER TABLE my_table ADD COLUMN processed_file_name varchar(256) NOT NULL DEFAULT '{file-name}';
这适用于我的用例,因为我在临时表上执行此操作,该表将仅保存一个文件中的数据,然后它们被截断并将数据传输到目标表。