shameet created FLINK-27189:
-------------------------------
Summary: Ability for table api to always add quotes to generated
csv
Key: FLINK-27189
URL: https://issues.apache.org/jira/browse/FLINK-27189
Project: Flink
Issue Type: Improvement
Components: API / Python
Reporter: shameet
I am using the table api in pyflink to generate a csv . What i noticed is that
its conditionally adding quotes around the data. What I want is quotes around
all the data
csv is being created in s3
e.g in output below the data in last column was not quoted
"[transaction_idgeorge.bl...@reqres.in|mailto:transaction_idgeorge.bl...@reqres.in]",card_hash,transaction_id
"[no3500957594177george.bl...@reqres.in|mailto:no3500957594177george.bl...@reqres.in]","bd9e27fdb53b114288948c18fdbf50e9931b935c30bd79ca288c50d3969e1ad5",NO3500957594177
"[no3500957594178george.bl...@reqres.in|mailto:no3500957594178george.bl...@reqres.in]","bd9e27fdb53b114288948c18fdbf50e9931b935c30bd79ca288c50d3969e1ad6",NO3500957594178
I had posted this question in user community and Dian FU suggested i could
create a Jira as this is not supported right now
[https://lists.apache.org/thread/y2g7kjf6ylmqtm2w9b28kfcdborvcgtn]
sample code to create a csv in s3
def create_source_table(table_name, input_path):
return""" CREATE TABLE \{0} (
transaction_id VARCHAR(100),
card_hash VARCHAR(100)
) with (
'connector' = 'filesystem',
'format' = 'csv',
'path' = '\{1}'
) """.format(
table_name, input_path)
def create_sink_table(table_name, bucket_name):
return""" CREATE TABLE \{0} (
transaction_id VARCHAR(100),
card_hash VARCHAR(100),
brand_id VARCHAR(100)
)
with (
'connector'='filesystem',
'path'='\{1}',
'format'='csv'
) """.format(
table_name, bucket_name)
2. Creates a source table from a Kinesis Data Stream
table_env.execute_sql(
create_source_table(
input_table_name, input_file
)
){color:#000000}
{color}
table_env.execute_sql(
create_sink_table(
out_table_name, output_bucket_name
)
){color:#000000}
{color}
table_env.register_function("addme1", addme1)
{color:#000000} {color}{color:#000000}
{color}{color:#000000}
{color}
source_table = table_env.from_path(input_table_name)
source_table.select(addme1(source_table.transaction_id),source_table.card_hash,
source_table.transaction_id.alias('brand_id')).execute_insert(out_table_name).wait()
{color:#202020} {color}
apache-flink version - 1.13
python 3.8
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
