zeroshade commented on issue #4548:
URL: https://github.com/apache/arrow-adbc/issues/4548#issuecomment-5026585708
If the table is a single chunk, then by default pyarrow will convert the
table into a single record batch and the driver will write it as is. If you
want the driver to write multiple files in parallel and upload them separately,
you need to also split the table as well. For example, by using `to_reader`:
```python
n = 80_000_000
data = pa.table(
{
"ID": pa.array(range(n), type=pa.int64()),
"VALUE": pa.array(np.random.randn(n), type=pa.float64()),
"CATEGORY": pa.array(["A", "B", "C", "D"] * (n // 4),
type=pa.string()),
}
)
db_kwargs = {
"adbc.snowflake.sql.account": "**redacted**",
"adbc.snowflake.sql.role": "**redacted**",
"username": "**redacted**",
"adbc.snowflake.sql.warehouse": "**redacted**_SMALL",
"adbc.snowflake.sql.auth_type": "auth_ext_browser",
"adbc.snowflake.sql.client_option.tracing": "debug",
}
print("Connecting...")
with adbc_snowflake.connect(db_kwargs=db_kwargs) as conn:
with conn.cursor() as cursor:
cursor.execute("USE DATABASE **redacted**")
cursor.execute("USE SCHEMA **redacted**")
# Try setting options on the statement
cursor.adbc_statement.set_options(
**{
"adbc.snowflake.statement.ingest_target_file_size":
"10485760",
"adbc.snowflake.statement.ingest_writer_concurrency": "4",
}
)
print("Ingesting...")
# rows = cursor.adbc_ingest("ADBC_SPLIT_TEST", data,
mode="create_append")
max_chunksize = 10_000_000
rows = cursor.adbc_ingest("ADBC_SPLIT_TEST",
data.to_reader(max_chunksize), mode="create_append")
print(f"Ingested {rows} rows")
conn.commit()
```
The above version using `to_reader` will split the table into 8 batches of
10M rows each and the driver will generate parquet files and upload them in
parallel with multiple PUT commands. The parallelism is configurable via
options like:
`adbc.snowflake.statement.ingest_writer_concurrency` --> number of
concurrent parquet file writers (default is NUM_CORES)
`adbc.snowflake.statement.ingest_upload_concurrency` --> number of
concurrent uploads (default is 8)
`adbc.snowflake.statement.ingest_copy_concurrency` --> number of concurrent
COPY INTO commands (default is 4)
You can also split into more chunks and set
`adbc.snowflake.statement.ingest_target_file_size` (default is 10MB) which will
verify the size of the parquet file after each written record batch to attempt
to limit the size. But because the driver currently only checks the size
*after* writing a batch, when a table is passed that is not chunked already it
will be sent as a single batch and written as a single file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
