From dd37227f2f1a13d6061ebe93dae5b38fc03537a4 Mon Sep 17 00:00:00 2001 From: Laurent Savaete Date: Sun, 4 Apr 2021 01:25:41 +0100 Subject: [PATCH 1/3] Cache _serialize_table_record_field_name to speedup loading Using a partial function and tweaking how the arguments are passed allows applying lru_cache on the method. The function is called for each field of each row, but the possible values for args are a lot more limited, so caching is very effective here. --- target_postgres/sql_base.py | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/target_postgres/sql_base.py b/target_postgres/sql_base.py index 9d879526..262e0cb9 100644 --- a/target_postgres/sql_base.py +++ b/target_postgres/sql_base.py @@ -13,6 +13,7 @@ # from copy import deepcopy +from functools import lru_cache, partial import time import singer @@ -628,16 +629,23 @@ def log_message(msg): return self._get_table_schema(connection, table_name) - def _serialize_table_record_field_name(self, remote_schema, path, value_json_schema): + def _serialize_table_record_field_name(self, remote_schema, path, value_json_schema_tuple): """ Returns the appropriate remote field (column) name for `path`. :param remote_schema: TABLE_SCHEMA(remote) :param path: (string, ...) - :value_json_schema: dict, JSON Schema + :value_json_schema: tuple, JSON Schema :return: string """ + # rebuild the dict that needs to be passed further down the call stack + if len(value_json_schema_tuple) == 1: + value_json_schema = { 'type': value_json_schema_tuple[0] } + else: + value_json_schema = {'type': value_json_schema_tuple[0], + 'format': value_json_schema_tuple[1]} + simple_json_schema = json_schema.simple_type(value_json_schema) mapping = self._get_mapping(remote_schema, @@ -725,6 +733,11 @@ def _serialize_table_records( default_row = dict([(field, NULL_DEFAULT) for field in remote_fields]) paths = streamed_schema['schema']['properties'].keys() + + ## create a partial function with only hashable args so we can use lru_cache on it + _cached_field_name = partial(self._serialize_table_record_field_name, remote_schema) + cached_field_name = lru_cache(maxsize=None)(_cached_field_name) + for record in records: row = deepcopy(default_row) @@ -747,18 +760,14 @@ def _serialize_table_records( and value is not None: value = self.serialize_table_record_datetime_value(remote_schema, streamed_schema, path, value) - value_json_schema = {'type': json_schema.STRING, - 'format': json_schema.DATE_TIME_FORMAT} + value_json_schema_tuple = (json_schema.STRING, json_schema.DATE_TIME_FORMAT) + field_name = cached_field_name(path, value_json_schema_tuple) else: - value_json_schema = {'type': json_schema_string_type} + field_name = cached_field_name(path, (json_schema_string_type,)) ## Serialize NULL default value value = self.serialize_table_record_null_value(remote_schema, streamed_schema, path, value) - field_name = self._serialize_table_record_field_name(remote_schema, - path, - value_json_schema) - ## `field_name` is unset if row[field_name] == NULL_DEFAULT: row[field_name] = value From ffc897661cb6dcb93148719ce08fea3473ce5832 Mon Sep 17 00:00:00 2001 From: Laurent Savaete Date: Sun, 4 Apr 2021 01:57:38 +0100 Subject: [PATCH 2/3] Cache datetime formatting function --- target_postgres/postgres.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/target_postgres/postgres.py b/target_postgres/postgres.py index 97d9ccf0..223c0d55 100644 --- a/target_postgres/postgres.py +++ b/target_postgres/postgres.py @@ -1,5 +1,6 @@ from copy import deepcopy import csv +from functools import lru_cache import io import json import logging @@ -19,6 +20,14 @@ RESERVED_NULL_DEFAULT = 'NULL' +@lru_cache(maxsize=128) +def _format_datetime(value): + """ + Format a datetime value. This is only called from the + PostgresTarget.serialize_table_record_datetime_value + but this non-method version allows caching + """ + return arrow.get(value).format('YYYY-MM-DD HH:mm:ss.SSSSZZ') def _update_schema_0_to_1(table_metadata, table_schema): """ @@ -546,7 +555,7 @@ def serialize_table_record_null_value(self, remote_schema, streamed_schema, fiel return value def serialize_table_record_datetime_value(self, remote_schema, streamed_schema, field, value): - return arrow.get(value).format('YYYY-MM-DD HH:mm:ss.SSSSZZ') + return _format_datetime(value) def persist_csv_rows(self, cur, From 7041e5c1352dc437e1c80598b08dbaf764bd4199 Mon Sep 17 00:00:00 2001 From: Laurent Savaete Date: Sun, 4 Apr 2021 01:58:11 +0100 Subject: [PATCH 3/3] Use pickle instead of deepcopy to speedup row copying --- target_postgres/sql_base.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/target_postgres/sql_base.py b/target_postgres/sql_base.py index 262e0cb9..2064005d 100644 --- a/target_postgres/sql_base.py +++ b/target_postgres/sql_base.py @@ -14,6 +14,7 @@ from copy import deepcopy from functools import lru_cache, partial +import pickle import time import singer @@ -740,7 +741,8 @@ def _serialize_table_records( for record in records: - row = deepcopy(default_row) + ## pickling/unpickling is much faster than deepcopy + row = pickle.loads(pickle.dumps(default_row)) for path in paths: json_schema_string_type, value = record.get(path, (None, None))