Repository navigation
BaseSQLToGCSOperator Parquet Format Fails to Write Dates/JSON #26248
Description
Apache Airflow Provider(s)
Versions of Apache Airflow Providers
8.3.0
Apache Airflow version
2.3.4
Operating System
OSX
Deployment
Virtualenv installation
Deployment details
No response
What happened
PostgresToGCSOperator was being used to query a table containing timestamps and JSON data.
export_format='parquet' was set.
The export to Parquet fails.
For the Timestamp, the error is:
File "/apache-airflow/airflow/providers/google/cloud/transfers/sql_to_gcs.py", line 154, in execute
for file_to_upload in self._write_local_data_files(cursor):
File "/apache-airflow/airflow/providers/google/cloud/transfers/sql_to_gcs.py", line 241, in _write_local_data_files
tbl = pa.Table.from_pydict(row_pydic, parquet_schema)
File "pyarrow/table.pxi", line 1724, in pyarrow.lib.Table.from_pydict
File "pyarrow/table.pxi", line 2385, in pyarrow.lib._from_pydict
File "pyarrow/array.pxi", line 341, in pyarrow.lib.asarray
File "pyarrow/array.pxi", line 315, in pyarrow.lib.array
File "pyarrow/array.pxi", line 39, in pyarrow.lib._sequence_to_array
File "pyarrow/error.pxi", line 143, in pyarrow.lib.pyarrow_internal_check_status
File "pyarrow/error.pxi", line 122, in pyarrow.lib.check_status
pyarrow.lib.ArrowTypeError: object of type <class 'str'> cannot be converted to int
For the JSON, the error is the same, except:
pyarrow.lib.ArrowTypeError: Expected bytes, got a 'dict' object
What you think should happen instead
Both timestamp and JSON datatypes should be supported for export to Parquet
I tested with JSON as the export format, and had no issues.
How to reproduce
As long as the export format is set to parquet, then selecting any timestamp or JSON object reproduces the issue.
I copied the affected part of code and built a stand-alone example that creates the same exception:
import pyarrow as pa
schema = pa.schema([
("current_sign_in_at", pa.date64()),
("registered_at", pa.date64())
])
py_row = {'current_sign_in_at': ['2019-02-06T15:11:46.991841'], 'registered_at': ['2019-02-06T15:14:18.626211']}
tbl = pa.Table.from_pydict(py_row, schema)
Anything else
While troubleshooting this, I believe the fix has to be added to every subclass of BaseSQLToGCSOperator. In the convert_type method, timestamps/dates/times are all converted to strings. This is good for CSV and JSON, but is causing issues with Parquet. Instead the types should be returned as-is.
As for JSON, probably just set stringify_dict=True for Parquet parsing is my guess.
Are you willing to submit PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct
Activity
zzsza commentedon Sep 11, 2022
I also had a similar error. It seems to be related to this issue
#17538
patricker commentedon Sep 12, 2022
@zzsza I agree, it's the same issue. I think this just confirms the fix will need to touch all of the child classes where Parquet is concerned.
potiuk commentedon Sep 18, 2022
I marked it as a good first issue for someone to solve (hopefully someone will be interested, pick it up and fix it) - but If you think you can open a PR @zzsza @patricker - opening a PR and leading it completions is the fastest way to get it fixed.
eladkal commentedon Sep 26, 2022
@akbarnotopb maybe it will be easier for you to start with this issue?
akbarnotopb commentedon Sep 26, 2022
yep i was having this issue earlier, the problem is well explained in this #17538 issue.
i can fix it later, but dont know how to make the pytest unit test working. *ive followed the guide with the vscode
im using docker-compose in this repo, will try to make the unit test working first. itll very helpful if you could tell me the easiest way to make it run 😁
eladkal commentedon Sep 26, 2022
You can join slack for support if you have difficulty
eladkal commentedon Dec 24, 2022
@vchiapaikeo maybe you can help with this issue as well? :)
vchiapaikeo commentedon Dec 26, 2022
Starting to take a look at this and at least for ints/dates, I can't seem to repro it. Trying w/ the below dag:
Task logs from the run:
And output from file seems fine:
I do think this repros with the json type though. Here's a log trying to export the serialized_dag table whose data object is of type json:
eladkal commentedon Dec 27, 2022
Thanks @vchiapaikeo !
Closing then as can't reproduce
vchiapaikeo commentedon Dec 27, 2022
Just to close the loop on the JSON issue - with the
stringify_dictoption set to True, this works just fine. The following dag exports successfully to GCS:And the output:
However, one of the drawbacks with this approach is that the JSON gets turned into a string type and not a deserialized representation which might be more efficient for querying parquet.
renzepost commentedon Jan 12, 2024
Hi,
I'm running into this very same issue. I'm using
MySQLToGCSOperatorwithexport_format="parquet".I'm on Cloud Composer 2.4.6, Airflow 2.6.3. I think as a workaround, I might either try exporting to CSV instead, or exporting the timestamp fields as Unix time.