25 lines
899 B
Python
25 lines
899 B
Python
 | 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()
|