SqlTableDataSet save_args dtype catalog definition

[scotthelm]

Hello! First off - this is the first issue I have yet to be able to find an answer for in the excellent kedro documentation. Kudos for that!

The project that I'm working on is a personal learning project to load world factbook data and make it quantifiable.

I have an issue where I am defining a SqlTableDataSet in my catalog. I want one of the columns to be a postgresql jsonb data type. I have defined the DataCatalog entry as:

silver_table_dataset:
  type: pandas.SQLTableDataSet
  table_name: factbook_data_silver
  save_args:
    if_exists: replace
    dtype:
      ethnic_groups: sqlalchemy.dialects.postgresql.json.JSONB
  credentials: dev_postgres_creds

In the dataframe, I have tried both a dict and json.dumps(dict) representation of the data. However, no matter the type of ethnic_groups nor the sqlalchemy type that I supply seems to work. The error that I receive is:

raise DataSetError(message) from exc
kedro.io.core.DataSetError: Failed while saving data to data set SQLTableDataSet(load_args={}, save_args={'dtype': {'ethnic_groups': sqlalchemy.dialects.postgresql.json.JSONB}, 'if_exists': replace, 'index': False}, table_name=factbook_data_silver).
The type of ethnic_groups is not a SQLAlchemy type

lib versions:

• kedro is 0.17.2
• sqlalchemy is 1.4.23
• pandas is 1.3.2

I would appreciate any help in understanding how to configure dtype in the catalog correctly.

Thanks!

[datajoely]

Hello does this dtype parameter suggested here work as part of your save_args?

https://stackoverflow.com/a/41469431

[datajoely]

Interesting - it would be helpful if you could test this using native pandas commands just to check if it's anything in our YAML config conversion which is stopping that working.

[scotthelm]

Using a kedro ipython session

In [1]: df = catalog.load('silver_csv_dataset')
2021-09-18 16:52:56,726 - kedro.io.data_catalog - INFO - Loading data from `silver_csv_dataset` (CSVDataSet)...

In [2]: creds = { 'con': 'postgresql://postgres:docker@db:5432/factbook_data' }

In [3]: df.to_sql(name='test_table', con=creds['con'], if_exists='replace', index=False, dtype={'e
   ...: thnic_groups': 'sqlalchemy.types.JSON'})

results in the same error. In digging a little further in that session, I played around with type TypeEngine and discovered that the string is not passing the "callable" check here and so it's returning the string, which is not an instance of TypeEngine. I'm not sure what the next step should be. My project is public, and available here. I may have something misconfigured such that pandas is not able to load types correctly.

[scotthelm]

I checked out this answer https://stackoverflow.com/questions/36755450/pandas-and-sql-alchemy-specify-column-data-types. I think the issue is that the pandas dtype is expecting to get the actual class and not the class name string. The following ipython session shows how I was able to have it load the column correctly:

In [1]: import sqlalchemy

In [2]: df = catalog.load('silver_csv_dataset')
2021-09-18 21:04:19,880 - kedro.io.data_catalog - INFO - Loading data from `silver_csv_dataset` (CSVDataSet)...

In [3]: creds = { 'con': 'postgresql://postgres:docker@db:5432/factbook_data' }

In [4]: df.to_sql(name='test_table', con=creds['con'], if_exists='replace', index=False, dtype={'e
   ...: thnic_groups': sqlalchemy.types.JSON})

In [5]: 

I am looking at how to use the postgresql specific jsonb field, but I think this is the root of the issue.

[scotthelm]

So to sum up the results, I am able to successfully make the pandas to_sql call by setting the dtype manually. So I wonder if there is an issue when using dtype in the save_args in the YAML config api?

[datajoely]

Hi @scotthelm - thank you for your digging here. I thought this may be the issue - Pandas is somewhat inconsistent here, sometimes it converts the strings to imports automatically (to_excel works like this IIRC).

The declarative YAML pattern in Kedro datasets means things like this are somewhat difficult to solve. The easiest quick and dirty fix is to subclass the pandas.SQLTableDataSet and make this work for your purposes.

As a Kedro team we could look to introduce some special syntax like:

silver_table_dataset:
  type: pandas.SQLTableDataSet
  table_name: factbook_data_silver
  save_args:
    if_exists: replace
    dtype:
      ethnic_groups: import(sqlalchemy.types.JSON)
  credentials: dev_postgres_creds

In terms of technical feasibility it feels very doable - however, I'm most hesitant about two things:

• Complicating the simple YAML syntax for something that can be solved by subclassing.
• Security risks allowing people to import anything at this point. I'm not too sure about this one - but it feels like something we should consider.

[datajoely]

Ultimately the current answer for this (whilst not perfect) is to subclass the pandas.SQLTableDataSet and handle type import yourself. As it currently stands the Kedro team will not introduce a special DSL for importing types within catalog entries, if this issue crops up more often it may be something we revisit.

[scotthelm]

I understand your reasoning, and you've laid out a path forward. Thanks very much for checking into it!

[scotthelm]

Hopefully this will help anyone watching this discussion. I was able to successfully subclass kedro.extras.datasets.pandas.SQLTableDataSet and allow dtype specified in the catalog to be respected by pandas. The solution that I came up with is here. The catalog entry that uses the subclass is here.

[scotthelm]

I may be premature. The datatype of the table is correct - it's a jsonb datatype, but pandas is importing it as an escaped string.

[scotthelm]

I have been able to solve the issue by ensuring the correct datatype of the column in my processing node. I believe the solution works.

[keua]

If this still interesting for someone, I solve the problem by using resolvers using kedro v0.19.7.

settings.py file:

from sqlalchemy.dialects.oracle import FLOAT
from sqlalchemy.types import Float
CONFIG_LOADER_ARGS = {
      "base_env": "base",
      "default_run_env": "local",
      "custom_resolvers": {
      "sqlalchemyfloat": lambda: Float(precision=53).with_variant(FLOAT(binary_precision=126), 'oracle')
    }

And then using in the catalog like this:

dtype:
      your_float_column: ${sqlalchemyfloat:}

[datajoely]

That's what it was added for!