| import os
|
| import pyarrow as pa
|
| import pyarrow.dataset as ds
|
| import deltalake as dl
|
|
|
| # Directory where Arrow/Parquet feature batches are written
|
| FEATURE_DIR = r"C:\DualEA_FeatureBatches"
|
| DELTA_DIR = r"C:\DualEA_DeltaLake"
|
| os.makedirs(DELTA_DIR, exist_ok=True)
|
|
|
| def ingest_to_delta():
|
| # Discover all Parquet files in the feature batch directory
|
| files = [os.path.join(FEATURE_DIR, f) for f in os.listdir(FEATURE_DIR) if f.endswith('.parquet')]
|
| if not files:
|
| print("No Parquet files found for ingestion.")
|
| return
|
| # Read all Parquet files into a single Arrow Table
|
| tables = [pa.parquet.read_table(f) for f in files]
|
| batch_table = pa.concat_tables(tables)
|
| # Write to Delta Lake
|
| dl.write_deltalake(DELTA_DIR, batch_table, mode="append")
|
| print(f"Ingested {len(files)} files to Delta Lake at {DELTA_DIR}")
|
|
|
| if __name__ == "__main__":
|
| ingest_to_delta()
|
