LinkedIn
Twitter
1 ACCEPTED SOLUTION
Hi @ms4446 , the issue I was facing before - java.lang.IllegalArgumentException: Failed to decode Schema due to an error decoding Field proto, was because of the varchar fields. I tried inserting integer data and it worked fine.
Now when I am trying with whole data by explicitly defining the schema, I am receiving following error message:
Caused by: java.lang.IllegalArgumentException: Unknown Coder URN beam:coder:pickled_python:v1. Known URNs: [beam:coder:avro:generic:v1, beam:coder:bytes:v1, beam:coder:bool:v1, beam:coder:string_utf8:v1, beam:coder:kv:v1, beam:coder:varint:v1, beam:coder:interval_window:v1, beam:coder:iterable:v1, beam:coder:timer:v1, beam:coder:length_prefix:v1, beam:coder:global_window:v1, beam:coder:windowed_value:v1, beam:coder:param_windowed_value:v1, beam:coder:double:v1, beam:coder:row:v1, beam:coder:sharded_key:v1, beam:coder:custom_window:v1, beam:coder:nullable:v1]
Following is the updated code, I am not sure, how to resolve this,
import argparse
import logging
import typing
import numpy as np
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.dataframe.io import read_csv
from apache_beam.dataframe.convert import to_pcollection
from apache_beam.io.jdbc import WriteToJdbc
import yaml
class Business(typing.NamedTuple):
series_reference: str
period: float
data_value: float
suppressed: str
status: str
units: str
magnitude: int
subject: str
group: str
series_title_1: str
series_title_2: str
series_title_3: str
series_title_4: str
series_title_5: str
def datatype_mapping(datatype):
if datatype == 'date':
return np.datetime64
elif datatype == 'varchar':
return str
elif datatype == 'bigint':
return np.int64
elif datatype == 'numeric':
return np.float64
def get_jdbc_url(host_ip, host_port, db_name):
return f'jdbc:postgresql://{host_ip}:{host_port}/{db_name}'
def get_table_details(config, source_system):
return config['database']['tableDetails'][source_system]['master_table_name']
def get_renamed_columns(original_columns):
columns_rename_map = dict()
for column in original_columns:
columns_rename_map[column] = column.lower()
return columns_rename_map
def run(argv=None):
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('--input', dest='input', default='gs://dti_source_data/business-financial-data-september-2023-quarter.csv', help='Input file to process')
parser.add_argument('--config_file', dest='config_file', help='Path to configuration file')
parser.add_argument('--source_system', dest='source_system', help='Source system name')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline = beam.Pipeline(options=PipelineOptions(pipeline_args))
with open(known_args.config_file, 'r') as f:
config = yaml.safe_load(f)
host_ip = config['database']['connectionDetails']['host_ip']
host_port = config['database']['connectionDetails']['host_port']
db_name = config['database']['connectionDetails']['db_name']
username = config['database']['connectionDetails']['username']
password = config['database']['connectionDetails']['password']
jdbc_url = get_jdbc_url(host_ip, host_port, db_name)
master_table = get_table_details(config, known_args.source_system)
columns_metadata_dict = config['database']['tableDetails'][known_args.source_system]['columns_metadata']
#DEBUG
logging.info(columns_metadata_dict)
dtype_dict = {}
parse_dates_list = []
for column in columns_metadata_dict:
if columns_metadata_dict[column] == 'date':
parse_dates_list.append(column)
else:
dtype_dict[column] = datatype_mapping(columns_metadata_dict[column])
# DEBUG
logging.info(dtype_dict)
logging.info(parse_dates_list)
with pipeline as p:
# Read the csv file as Deferred dataframe using Apache Beam native Dataframe
deferred_dataframe = p | read_csv(known_args.input, delimiter=',', dtype=dtype_dict, quotechar='"', parse_dates = parse_dates_list, na_values='')
# Rename the columns to lower case names to comply with Cloud SQL for PostgreSQL table
columns_rename_map = get_renamed_columns(deferred_dataframe.columns)
deferred_dataframe.rename(columns=columns_rename_map, inplace=True)
# DEBUG
logging.info(deferred_dataframe.dtypes)
dataframe_to_pcollection = to_pcollection(deferred_dataframe)
required_column_dict = config['database']['tableDetails'][known_args.source_system]['columns_metadata']
required_column_list = []
for column_name in required_column_dict:
required_column_list.append(column_name.lower())
filtered_pcollection = dataframe_to_pcollection | beam.Select(*required_column_list)
output_pcollection = filtered_pcollection | beam.Map(lambda element: element).with_output_types(Business)
output_pcollection | beam.io.WriteToCsv(path='gs://dti_source_data/processed.csv')
output_pcollection | WriteToJdbc(
table_name=master_table,
driver_class_name='org.postgresql.Driver',
jdbc_url=jdbc_url,
username=username,
password=password
)
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()The target table schema is as follows:
Hi @ms4446 , I hope you are doing great.
import argparse
import logging
import typing
import numpy as np
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.dataframe.io import read_csv
from apache_beam.dataframe.convert import to_pcollection
from apache_beam import coders
import yaml
def datatype_mapping(datatype):
if datatype == 'date':
return np.datetime64
elif datatype == 'varchar':
return object
elif datatype == 'bigint':
return np.int64
elif datatype == 'numeric':
return np.float64
def get_jdbc_url(host_ip, host_port, db_name):
return f'jdbc:postgresql://{host_ip}:{host_port}/{db_name}'
def get_table_details(config, source_system):
return config['database']['tableDetails'][source_system]['master_table_name']
def get_renamed_columns(original_columns):
columns_rename_map = dict()
for column in original_columns:
columns_rename_map[column] = column.lower()
return columns_rename_map
def get_connection_string(db_name, db_user, db_pass, host_ip, host_port):
connection_string = f'dbname={db_name} user={db_user} password={db_pass} host={host_ip} port={host_port}'
return connection_string
class Business(typing.NamedTuple):
series_reference: typing.Optional[str]
period: typing.Optional[float]
data_value: typing.Optional[float]
suppressed: typing.Optional[str]
status: typing.Optional[str]
units: typing.Optional[str]
magnitude: typing.Optional[int]
subject: typing.Optional[str]
group: typing.Optional[str]
series_title_1: typing.Optional[str]
series_title_2: typing.Optional[str]
series_title_3: typing.Optional[str]
series_title_4: typing.Optional[str]
series_title_5: typing.Optional[str]
coders.registry.register_coder(Business, coders.RowCoder)
class ConvertToKeyValuePairs(beam.DoFn):
def process(self, element):
argslist = list()
for key, value in element.__dict__.items():
argslist.append((key, value))
yield (element.__dict__['series_reference'], tuple(argslist))
class WriteToDB(beam.DoFn):
def __init__(self, connection_string, table_name):
self.connection_string = connection_string
self.table_name = table_name
def process(self, element):
import psycopg2
import numpy as np
from psycopg2.extras import execute_batch
from psycopg2.extensions import register_adapter, AsIs
register_adapter(np.int64, AsIs)
conn = psycopg2.connect(self.connection_string)
with conn.cursor() as cur:
columns = []
values = []
for row in element[1]:
for column in row:
columns.append(column[0])
break
for row in element[1]:
temp_values = []
for column in row:
temp_values.append(column[1])
values.append(tuple(temp_values))
placeholders = ','.join(['%s'] * len(columns))
columns_string = '"' + '","'.join(columns) + '"'
insert_query = f'INSERT INTO {self.table_name} ({columns_string}) VALUES ({placeholders})'
execute_batch(cur, insert_query, tuple(values))
conn.commit()
conn.close()
def run(argv=None):
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('--input', dest='input', default='gs://dti_source_data/business-financial-data-september-2023-quarter.csv', help='Input file to process')
parser.add_argument('--config_file', dest='config_file', help='Path to configuration file')
parser.add_argument('--source_system', dest='source_system', help='Source system name')
parser.add_argument('--key_columns', dest='key_columns', help='Key columns string separated by comma')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline = beam.Pipeline(options=PipelineOptions(pipeline_args))
with open(known_args.config_file, 'r') as f:
config = yaml.safe_load(f)
host_ip = config['database']['connectionDetails']['host_ip']
host_port = config['database']['connectionDetails']['host_port']
db_name = config['database']['connectionDetails']['db_name']
username = config['database']['connectionDetails']['username']
password = config['database']['connectionDetails']['password']
psycopg_connection_string = get_connection_string(db_name, username, password, host_ip, host_port)
jdbc_url = get_jdbc_url(host_ip, host_port, db_name)
master_table = get_table_details(config, known_args.source_system)
columns_metadata_dict = config['database']['tableDetails'][known_args.source_system]['columns_metadata']
#DEBUG
logging.info(columns_metadata_dict)
dtype_dict = {}
parse_dates_list = []
for column in columns_metadata_dict:
if columns_metadata_dict[column] == 'date':
parse_dates_list.append(column)
else:
dtype_dict[column] = datatype_mapping(columns_metadata_dict[column])
# DEBUG
logging.info(dtype_dict)
logging.info(parse_dates_list)
with pipeline as p:
# Read the csv file as Deferred dataframe using Apache Beam native Dataframe
deferred_dataframe = p | "Read the CSV files from GCS" >> read_csv(known_args.input, delimiter=',', dtype=dtype_dict, quotechar='"', parse_dates = parse_dates_list, na_values='')
# Rename the columns to lower case names to comply with Cloud SQL for PostgreSQL table
columns_rename_map = get_renamed_columns(deferred_dataframe.columns)
deferred_dataframe.rename(columns=columns_rename_map, inplace=True)
# DEBUG
logging.info(deferred_dataframe.dtypes)
dataframe_to_pcollection = to_pcollection(deferred_dataframe)
required_column_dict = config['database']['tableDetails'][known_args.source_system]['columns_metadata']
required_column_list = []
for column_name in required_column_dict:
required_column_list.append(column_name.lower())
required_columns_pcollection = dataframe_to_pcollection | "Filter the PCollection" >> beam.Select(*required_column_list)
output_pcollection = required_columns_pcollection | "Map to Business" >> beam.Map(lambda element: element).with_output_types(Business)
# output_pcollection | "Write PCollection to GCS as CSV file" >> beam.io.WriteToCsv(path='gs://dti_source_data/processed.csv')
key_value_pairs_pcollection = output_pcollection | beam.ParDo(ConvertToKeyValuePairs())
batched_pcollection = key_value_pairs_pcollection | beam.GroupIntoBatches(batch_size=200)
batched_pcollection | beam.ParDo(WriteToDB(psycopg_connection_string, master_table))
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()Above is my current code, but the problem I am facing here is, when I run this pipeline for my actual data which is approximately 288 MB (450000 records) for one of the source system, it takes an hour to complete, and the write IO throughput is no going above approx. 180/s. Please check below screenshot.
Can you please guide me, how can I increase the Write IO, and increase the pipeline performance.
Further, I also observed that, the CPU and Memory are not getting utilized, instead it auto scales because the steps progress slow
