(df, schema, preserve_index, nthreads=1, columns=None,
safe=True)
| 597 | |
| 598 | |
| 599 | def dataframe_to_arrays(df, schema, preserve_index, nthreads=1, columns=None, |
| 600 | safe=True): |
| 601 | (all_names, |
| 602 | column_names, |
| 603 | column_field_names, |
| 604 | index_column_names, |
| 605 | index_descriptors, |
| 606 | index_columns, |
| 607 | columns_to_convert, |
| 608 | convert_fields) = _get_columns_to_convert(df, schema, preserve_index, |
| 609 | columns) |
| 610 | |
| 611 | # NOTE(wesm): If nthreads=None, then we use a heuristic to decide whether |
| 612 | # using a thread pool is worth it. Currently the heuristic is whether the |
| 613 | # nrows > 100 * ncols and ncols > 1. |
| 614 | if nthreads is None: |
| 615 | nrows, ncols = len(df), len(df.columns) |
| 616 | if nrows > ncols * 100 and ncols > 1: |
| 617 | nthreads = pa.cpu_count() |
| 618 | else: |
| 619 | nthreads = 1 |
| 620 | # if we don't have threading in libarrow, don't use threading here either |
| 621 | if not is_threading_enabled(): |
| 622 | nthreads = 1 |
| 623 | |
| 624 | def convert_column(col, field): |
| 625 | if field is None: |
| 626 | field_nullable = True |
| 627 | type_ = None |
| 628 | else: |
| 629 | field_nullable = field.nullable |
| 630 | type_ = field.type |
| 631 | |
| 632 | try: |
| 633 | result = pa.array(col, type=type_, from_pandas=True, safe=safe) |
| 634 | except (pa.ArrowInvalid, |
| 635 | pa.ArrowNotImplementedError, |
| 636 | pa.ArrowTypeError) as e: |
| 637 | e.args += ( |
| 638 | f"Conversion failed for column {col.name} with type {col.dtype}",) |
| 639 | raise e |
| 640 | if not field_nullable and result.null_count > 0: |
| 641 | raise ValueError(f"Field {field} was non-nullable but pandas column " |
| 642 | f"had {result.null_count} null values") |
| 643 | return result |
| 644 | |
| 645 | def _can_definitely_zero_copy(arr): |
| 646 | return (isinstance(arr, np.ndarray) and |
| 647 | arr.flags.contiguous and |
| 648 | issubclass(arr.dtype.type, np.integer)) |
| 649 | |
| 650 | if nthreads == 1: |
| 651 | arrays = [convert_column(c, f) |
| 652 | for c, f in zip(columns_to_convert, convert_fields)] |
| 653 | else: |
| 654 | arrays = [] |
| 655 | with futures.ThreadPoolExecutor(nthreads) as executor: |
| 656 | for c, f in zip(columns_to_convert, convert_fields): |
nothing calls this directly
no test coverage detected