crowemi/target-s3

singer.io target for S3 built with @meltano SDK

target-s3

target-s3 is inteded to be a multi-format/multi-cloud Singer target.

Build with the Meltano Target SDK.

Configuration

Accepted Config Options

{
    "format": {
        "format_type": "json",
        "format_parquet": {
            "validate": true|false
        },
        "format_json": {},
        "format_csv": {}
    },
    "cloud_provider": {
        "cloud_provider_type": "aws",
        "aws": {
            "aws_access_key_id": "test",
            "aws_secret_access_key": "test",
            "aws_region": "us-west-2",
            "aws_profile_name": "test-profile",
            "aws_bucket": "test-bucket",
            "aws_endpoint_override": "http://localhost:4566"
        }
    },
    "prefix": "path/to/output",
    "stream_name_path_override": "StreamName",
    "include_process_date": true|false,
    "append_date_to_prefix": true|false,
    "partition_name_enabled": true|false,
    "use_raw_stream_name": true|false,
    "append_date_to_prefix_grain": "day",
    "append_date_to_filename": true|false,
    "append_date_to_filename_grain": "microsecond",
    "flattening_enabled": true|false,
    "flattening_max_depth": int,
    "max_batch_age": int,
    "max_batch_size": int
}

format.format_parquet.validate [Boolean, default: False] - this flag determines whether the data types of incoming data elements should be validated. When set True, a schema is created from the first record and all subsequent records that don't match that data type are cast.

Capabilities

• about
• stream-maps
• schema-flattening

Settings

Configure using environment variables

This Singer target will automatically import any environment variables within the working directory's .env if the --config=ENV is provided, such that config values will be considered if a matching environment variable is set either in the terminal context or in the .env file.

Usage

You can easily run target-s3 by itself or in a pipeline using Meltano.

Executing the Target Directly

target-s3 --version
target-s3 --help
# Test using the "Carbon Intensity" sample:
tap-carbon-intensity | target-s3 --config /path/to/target-s3-config.json

SDK Dev Guide

See the dev guide for more instructions on how to use the Meltano Singer SDK to develop your own Singer taps and targets.

target-s3's Issues

Empty file name when append_date_to_filename=False

Hello,

I'm using target-s3 within Meltano and want to overwrite the Parquet files I'm creating with every run.

I set the TARGET_S3_APPEND_DATE_TO_FILENAME and TARGET_S3_APPEND_DATE_TO_PREFIX to False but now I end up with files in S3 like so: bucket_name/prefix/stream_name/.parquet

It is still possible to query this file but feels a bit odd to me.

As you can see here the filename is only set when append_date_to_filename is True.

I would expect that when both append_date_to_filename and append_date_to_prefix is set to false the filename would be equal to the stream_name resulting in this: bucket_name/prefix/stream_name.parquet

Is there any reason behind it?

I could make a PR adjusting this if it is welcome.

bug: `PermanentRedirect Message`

I'm getting the following error when I try to use the parquet format type:

AWS Error UNKNOWN (HTTP status 301) during CreateMultipartUpload operation: Unable to parse ExceptionName: PermanentRedirect Message: The bucket you are attempting to access must be addressed using the specified endpoint. Please send all future requests to this endpoint.

It looks like in this commit the logic for creating a file system changed. If I put it back to what it was I'm able to get it to work. I think the error is related to the region not properly being set

Also if I'm reading it correctly I think self.session gets instantiated on init each time so its never possible to get to

target-s3/target_s3/formats/format_parquet.py

Line 30 in 6dcf59f

else:

, is that right?

target-s3 always appends seconds to path prefix if grain is hours or lower

I don't know if it's intentional but:

target-s3/target_s3/formats/format_base.py

Line 135 in 709734e

ret += f"{'second=' if partition_name_enabled else ''}{batch_start.second:02}/" if grain <= 4 else ""

Means that seconds are always added when the grain is hour or lower

Format jsonl not working

Hi, tried setting the format to jsonl like so:

format:
  format_type: jsonl
  format_parquet:
    validate: false

Got immediate error:
'jsonl' is not a valid choice for 'format.format_type'

Using Meltano Docker image latest

Thanks for all your hard work!!!

config credentials not used yet

I noticed that I can set credentials in the config but they arent used anywhere yet. Setting them in the environment does seem to work.

Support MAX_BATCH_SIZE env variable

The TARGET_S3_MAX_BATCH_SIZE env var only seems to work whenever I declare my own default max_batch_size value in the meltano.yml configuration. This doesn't seem to be the case for other environment variables such as TARGET_S3_APPEND_DATE_TO_FILENAME_GRAIN.

I'm guessing this is also related to how max_batch_size doesn't appear on the meltano hub documentation page or in the meltano config target-s3 set --interactive list.

Is this expected behavior? I'm on Meltano 2.20.0 for... reasons.

Supporting different pyarrow filesystems

Hey nice target!

Any interest in contributions to make this support other filesystems like gcs?

Add a way to pass `endpoint_url` to target-s3

We are attempting to use target-s3 in a pipeline as it is capable of creating parquet files.

However as part of our development and QA processes we must to have a way of testing the pipeline. For this we would like to use either MinIO or localstack but this is currently not possible as target-s3 has no config for setting the endpoint_url.

Requested solution

Add endpoint_url config to work with the internal boto3 and pyarrow settings.

Many thanks.

feat: implement csv writer

The csv format type logic still needs to be written. I think the current base format and smart open handle most of this for us already but we need to figure out the best way to pass in the csv config options for common use cases e.g. delimter, quotechar, encoding, etc.

using SDK flattening logic in the plugin.

Using https://github.com/meltano/sdk/blob/main/singer_sdk/helpers/_flattening.py to do the flattening in the plugin.

adding aws_session_token to config

To access AWS it is sometimes requested to use an aws_session_token as well.

I forked:
Vzzarr@25e556a

aws_session_token=aws_config.get("aws_session_token", None),

and this is working for me.
Would you please consider adding this to the code base or allow me to create a branch to PR?

parquet schema wrongly inferred

if there's a nested object that is sparse in nature, meaning occurs infrequently in data. Then for that data the inferred schema would be wrong.

eg.
correct schema: ["attributes"]["teams"]["assigned"] = list<item: struct<email: string>>>
but if there's no data for the above attributes, due to how schema is currently inferred here the schema becomes:
list<item: null> which can be tricky if the files are then being loaded to spark for downstream consumption.

I see there's a comment already to build schema from json schema rather than inferring it which I believe is the right way.

Just opening this issue, so I can pick it up later.

OMG RAM Usage!

Hi, probably known issue but.. wow that's insane RAM usage! 🤣

Reading the default 10K rows per batch it consumes 4 to 12 GB of memory in my case. Amazing!

Would love to trace the issue, hope I'll find the time.
So here's a place to discuss and track it.
Any pointers / comments / observations most welcome.

Thanks!

File is overwritten with each sink operation

When using target-s3 with over 10k records, we've noticed that the only way to achieve the desired outcome is by using the "append_date_to_prefix_grain": "microsecond" option. This is due to the fact that each sink operation overwrites the generated file instead of appending the data. Consequently, if more than 10K data is sent by the tap to the target within a short time frame, some data may be lost during the file writing process. Could we consider modifying the command from "w" to "a"? Is there a particular reason for using the write operation? If so, it might be necessary to create a new file for each sink operation to prevent data loss.

The target-s3-parquet uses append mode to write data.
https://github.com/gupy-io/target-s3-parquet/blob/main/target_s3_parquet/sinks.py#L83C15-L83C25

collections.Mutable python3.10 support

Currently imports for collections.Mutable is not updated for Python 3.10. I'm more than happy to put forth a PR, or someone could just apply the import changes in formats/format_base.py

--- a/target_s3/formats/format_base.py                                                                                                                                                                                 
+++ b/target_s3/formats/format_base.py               
@@ -1,7 +1,6 @@
 import re
 import inflection
 import json
-import collections
 import logging
 from datetime import datetime
 from abc import ABCMeta, abstractmethod             
@@ -9,6 +8,11 @@ from abc import ABCMeta, abstractmethod
 from boto3 import Session
 from smart_open import open
  
+try:
+    from collections.abc import MutableMapping      
+except ImportError
+    from collections import MutableMapping
+
  
 LOGGER = logging.getLogger("target-s3")                                                                                                                                                                               
 DATE_GRAIN = {                                                                                                                                                                                                                                                                                                                             
@@ -175,7 +179,7 @@ class FormatBase(metaclass=ABCMeta):                                                                                                                                                               
         for k in sorted(d.keys()):                                                                                                                                                                                    
             v = d[k]                                                                                                                                                                                                  
             new_key = self.flatten_key(k, parent_key, sep)                                                                                                                                                            
-            if isinstance(v, collections.MutableMapping):                                                                                                                                                             
+            if isinstance(v, MutableMapping):                                                                                                                                                                         
                 items.extend(self.flatten_record(v, parent_key + [k], sep=sep).items())                                                                                                                               
             else:                                                                                                                                                                                                     
                 items.append((new_key, json.dumps(v) if type(v) is list else v))  

bump sdk version to 0.32

working with older versions is a bit difficult especially with the flattening logic failing for some taps/streams. Is this in focus?

bug: `AttributeError: 'FormatParquet' object has no attribute 'file_iterator'`

When using the parquet type I get 1 successful file written to S3 but then it fails due to an AttributeError. Here is the full traceback:

2023-04-03 13:39:34,027 Target 'target-s3' is listening for input from tap.
2023-04-03 13:39:34,027 Initializing 'target-s3' target sink...
2023-04-03 13:39:34,027 Initializing target sink for stream 'Account'...
2023-04-03 13:39:34,028 Target 'target-s3' completed reading 4 lines of input (2 records, (0 batch manifests, 1 state messages).
2023-04-03 13:39:34,040 key: <my-bucket>/s3_loader/Account/2023/04/03/20230403
2023-04-03 13:39:35,485 Failed to write parquet file to S3.
Traceback (most recent call last):
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/bin/target-s3", line 8, in <module>
    sys.exit(Targets3.cli())
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 1130, in __call__
    return self.main(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 1055, in main
    rv = self.invoke(ctx)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 760, in invoke
    return __callback(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 564, in cli
    target.listen(file_input)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/io_base.py", line 35, in listen
    self._process_endofpipe()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 280, in _process_endofpipe
    self.drain_all(is_endofpipe=True)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 441, in drain_all
    self._drain_all(list(self._sinks_active.values()), self.max_parallelism)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 474, in _drain_all
    Parallel()(delayed(_drain_sink)(sink=sink) for sink in sink_list)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 1098, in __call__
    self.retrieve()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 975, in retrieve
    self._output.extend(job.get(timeout=self.timeout))
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/multiprocessing/pool.py", line 771, in get
    raise self._value
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/multiprocessing/pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/_parallel_backends.py", line 620, in __call__
    return self.func(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 288, in __call__
    return [func(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 288, in <listcomp>
    return [func(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 471, in _drain_sink
    self.drain_one(sink)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 461, in drain_one
    sink.process_batch(draining_status)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/sinks.py", line 48, in process_batch
    format_type_client.run()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_parquet.py", line 57, in run
    return super().run(self.context['records'])
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_base.py", line 62, in run
    self._write()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_parquet.py", line 53, in _write
    raise e
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_parquet.py", line 50, in _write
    self.file_iterator += 1
AttributeError: 'FormatParquet' object has no attribute 'file_iterator'

TypeError: Object of type datetime is not JSON serializable

This is related to #1, which is marked as done. However, it is still reproducible in the latest version for schema where fields contain date-time format, for example

"created_on": {
      "type": "string",
      "format": "date-time",
      "description": "The date and time when the repository was created."
},

Here is the error message

2024-06-19T14:26:11.471426Z [info     ] joblib.externals.loky.process_executor._RemoteTraceback: cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.471685Z [info     ] """                            cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.471858Z [info     ] Traceback (most recent call last): cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472008Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/_utils.py", line 72, in __call__ cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472176Z [info     ]     return self.func(**kwargs) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472321Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 598, in __call__ cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472472Z [info     ]     return [func(*args, **kwargs) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472615Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 598, in <listcomp> cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472756Z [info     ]     return [func(*args, **kwargs) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.472894Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/target_base.py", line 509, in _drain_sink cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.473034Z [info     ]     self.drain_one(sink)       cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.473174Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/target_base.py", line 499, in drain_one cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.473364Z [info     ]     sink.process_batch(draining_status) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.473571Z [info     ]   File "/home/user/dev/target-s3/target_s3/sinks.py", line 65, in process_batch cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.473773Z [info     ]     format_type_client.run()   cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.473972Z [info     ]   File "/home/user/dev/target-s3/target_s3/formats/format_jsonl.py", line 34, in run cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.474112Z [info     ]     return super().run(self.context["records"]) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.474673Z [info     ]   File "/home/user/dev/target-s3/target_s3/formats/format_base.py", line 99, in run cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.474957Z [info     ]     self._write()              cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.475217Z [info     ]   File "/home/user/dev/target-s3/target_s3/formats/format_jsonl.py", line 30, in _write cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.475412Z [info     ]     return super()._write('\n'.join(map(dumps, self.records))) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.475560Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/simplejson/__init__.py", line 378, in dumps cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.475753Z [info     ]     return _default_encoder.encode(obj) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.475968Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/simplejson/encoder.py", line 298, in encode cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.476217Z [info     ]     chunks = self.iterencode(o) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.476462Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/simplejson/encoder.py", line 379, in iterencode cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.476732Z [info     ]     return _iterencode(o, 0)   cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.477085Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/simplejson/encoder.py", line 274, in default cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.477409Z [info     ]     raise TypeError('Object of type %s is not JSON serializable' % cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.477671Z [info     ] TypeError: Object of type datetime is not JSON serializable cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.477929Z [info     ] """                            cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.478151Z [info     ]                                cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.478328Z [info     ] The above exception was the direct cause of the following exception: cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.478556Z [info     ]                                cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.478796Z [info     ] Traceback (most recent call last): cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.479032Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/bin/target-s3", line 8, in <module> cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.479366Z [info     ]     sys.exit(Targets3.cli())   cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.479614Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/click/core.py", line 1157, in __call__ cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.479785Z [info     ]     return self.main(*args, **kwargs) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.480044Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/click/core.py", line 1078, in main cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.480220Z [info     ]     rv = self.invoke(ctx)      cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.480373Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/click/core.py", line 1434, in invoke cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.480588Z [info     ]     return ctx.invoke(self.callback, **ctx.params) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.480799Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/click/core.py", line 783, in invoke cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.483655Z [info     ]     return __callback(*args, **kwargs) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.483901Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/target_base.py", line 554, in invoke cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.484137Z [info     ]     target.listen(file_input)  cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.484352Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/io_base.py", line 35, in listen cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.484573Z [info     ]     self._process_endofpipe()  cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.484781Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/target_base.py", line 314, in _process_endofpipe cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.485002Z [info     ]     self.drain_all(is_endofpipe=True) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.485215Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/target_base.py", line 479, in drain_all cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.485430Z [info     ]     self._drain_all(list(self._sinks_active.values()), self.max_parallelism) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.485631Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/singer_sdk/target_base.py", line 512, in _drain_all cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.485828Z [info     ]     Parallel()(delayed(_drain_sink)(sink=sink) for sink in sink_list) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.486022Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 2007, in __call__ cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.486219Z [info     ]     return output if self.return_generator else list(output) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.486413Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 1650, in _get_outputs cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.486607Z [info     ]     yield from self._retrieve() cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.486801Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 1754, in _retrieve cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.486995Z [info     ]     self._raise_error_fast()   cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.487190Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 1789, in _raise_error_fast cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.487383Z [info     ]     error_job.get_result(self.timeout) cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.487575Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 745, in get_result cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.487767Z [info     ]     return self._return_or_raise() cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.487965Z [info     ]   File "/home/user/dev/tap-bitbucket/.meltano/loaders/target-s3/venv/lib/python3.10/site-packages/joblib/parallel.py", line 763, in _return_or_raise cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.488857Z [info     ]     raise self._result         cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.489116Z [info     ] TypeError: Object of type datetime is not JSON serializable cmd_type=loader name=target-s3 run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.681247Z [error    ] Loading failed                 code=1 message=TypeError: Object of type datetime is not JSON serializable name=meltano run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3
2024-06-19T14:26:11.682439Z [info     ] ELT could not be completed: Loader failed. cmd_type=elt name=meltano run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.682844Z [info     ] For more detailed log messages re-run the command using 'meltano --log-level=debug ...' CLI flag. cmd_type=elt name=meltano run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.683057Z [info     ] Note that you can also check the generated log file at '/home/user/dev/tap-bitbucket/.meltano/logs/elt/2024-06-19T142543--tap-bitbucket--target-s3/4f8d59d3-b929-4f5c-8b62-3a2d6446f63a/elt.log'. cmd_type=elt name=meltano run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr
2024-06-19T14:26:11.683259Z [info     ] For more information on debugging and logging: https://docs.meltano.com/reference/command-line-interface#debugging cmd_type=elt name=meltano run_id=4f8d59d3-b929-4f5c-8b62-3a2d6446f63a state_id=2024-06-19T142543--tap-bitbucket--target-s3 stdio=stderr

Removing format, and keeping it as sting works.

bug: `TypeError: Object of type datetime is not JSON serializable`

I'm getting an error with format_type: json when theres a datetime value in my records.

2023-04-03 13:40:43,236 Target 'target-s3' is listening for input from tap.
2023-04-03 13:40:43,236 Initializing 'target-s3' target sink...
2023-04-03 13:40:43,236 Initializing target sink for stream 'Account'...
2023-04-03 13:40:43,237 Target 'target-s3' completed reading 4 lines of input (2 records, (0 batch manifests, 1 state messages).
2023-04-03 13:40:43,249 key: <my-bucket>/s3_loader/Account/2023/04/03/20230403
Traceback (most recent call last):
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/bin/target-s3", line 8, in <module>
    sys.exit(Targets3.cli())
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 1130, in __call__
    return self.main(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 1055, in main
    rv = self.invoke(ctx)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/click/core.py", line 760, in invoke
    return __callback(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 564, in cli
    target.listen(file_input)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/io_base.py", line 35, in listen
    self._process_endofpipe()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 280, in _process_endofpipe
    self.drain_all(is_endofpipe=True)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 441, in drain_all
    self._drain_all(list(self._sinks_active.values()), self.max_parallelism)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 474, in _drain_all
    Parallel()(delayed(_drain_sink)(sink=sink) for sink in sink_list)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 1098, in __call__
    self.retrieve()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 975, in retrieve
    self._output.extend(job.get(timeout=self.timeout))
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/multiprocessing/pool.py", line 771, in get
    raise self._value
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/multiprocessing/pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/_parallel_backends.py", line 620, in __call__
    return self.func(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 288, in __call__
    return [func(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/joblib/parallel.py", line 288, in <listcomp>
    return [func(*args, **kwargs)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 471, in _drain_sink
    self.drain_one(sink)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/singer_sdk/target_base.py", line 461, in drain_one
    sink.process_batch(draining_status)
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/sinks.py", line 48, in process_batch
    format_type_client.run()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_json.py", line 20, in run
    return super().run(self.context['records'])
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_base.py", line 62, in run
    self._write()
  File "/Users/pnadolny/Documents/Git/meltano_project/quickbooks_postgres/qb_pg/.meltano/loaders/target-s3/venv/lib/python3.9/site-packages/target_s3/formats/format_json.py", line 16, in _write
    return super()._write(json.dumps(self.records))
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/json/__init__.py", line 231, in dumps
    return _default_encoder.encode(obj)
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/json/encoder.py", line 199, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/json/encoder.py", line 257, in iterencode
    return _iterencode(o, 0)
  File "/Users/pnadolny/.pyenv/versions/3.9.7/lib/python3.9/json/encoder.py", line 179, in default
    raise TypeError(f'Object of type {o.__class__.__name__} '
TypeError: Object of type datetime is not JSON serializable

Configure Compression Settings

I have tried target-s3 and I was able to change prefix and stream_name_path_override but inside the S3 folder I get a compressed ".json.gz". Instead, I need a file called "data.json" without compression.

Partition formatting structure

Still me bothering you.

I tried to use this config

    "append_date_to_prefix": true,
    "append_date_to_prefix_grain": "day",

but the problem with this is that it creates partitions in the form of year/month/day and not year=year/month=month/day=day, for example: 2023/05/01 instead of year=2023/month=05/day=01, which is the idea day when scanning this data with other services like Redshift Spectrum.

Am I using wrongly the library and there is a parameter I'm missing or is this something that can be improved?

Create automated unit tests

Create automated unit tests which cover the following objects:

• format_base
• format_parquet
• format_json

Tests should run as part of GitHub Actions and successful tests should be a requirement for merge.

Define/validate parquet schema

We need a way to explicitly define and validate the parquet schema based on the incoming records. PyArrow creates an implied schema, but doesn't gracefully handle dynamic schemas.

https://arrow.apache.org/docs/python/generated/pyarrow.Schema.html