|
|
|
|
|
from pyspark.sql import SparkSession |
|
|
from pyspark.sql.types import StructType, StructField, LongType, DoubleType, IntegerType |
|
|
from pyspark.sql.functions import col, row_number, floor, first, max, min, last, sum |
|
|
from pyspark.sql.window import Window |
|
|
import os |
|
|
import sys |
|
|
import shutil |
|
|
import pandas as pd |
|
|
|
|
|
|
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) |
|
|
from minio_api.client import sign_in |
|
|
from minio_api.minio_utils import get_minio_data |
|
|
|
|
|
def initialize_spark_session(app_name="MinIO to Spark DataFrame", |
|
|
driver_memory="4g", executor_memory="4g"): |
|
|
return SparkSession.builder \ |
|
|
.appName(app_name) \ |
|
|
.config("spark.driver.memory", driver_memory) \ |
|
|
.config("spark.executor.memory", executor_memory) \ |
|
|
.getOrCreate() |
|
|
|
|
|
def create_dataframe_from_csv(spark, csv_lines, temp_parquet_path="temp/temp_parquet_chunks", |
|
|
chunk_size=int(3e+6)): |
|
|
os.makedirs(temp_parquet_path, exist_ok=True) |
|
|
schema = StructType([ |
|
|
StructField("Open time", LongType(), True), |
|
|
StructField("Open", DoubleType(), True), |
|
|
StructField("High", DoubleType(), True), |
|
|
StructField("Low", DoubleType(), True), |
|
|
StructField("Close", DoubleType(), True), |
|
|
StructField("Volume", DoubleType(), True), |
|
|
StructField("Close time", LongType(), True), |
|
|
StructField("Quote asset volume", DoubleType(), True), |
|
|
StructField("Number of trades", IntegerType(), True), |
|
|
StructField("Taker buy base asset volume", DoubleType(), True), |
|
|
StructField("Taker buy quote asset volume", DoubleType(), True), |
|
|
StructField("Ignore", IntegerType(), True) |
|
|
]) |
|
|
|
|
|
if csv_lines and csv_lines[0].startswith("Open time,"): |
|
|
data_lines = csv_lines[1:] |
|
|
else: |
|
|
data_lines = csv_lines |
|
|
|
|
|
if os.path.exists(temp_parquet_path): |
|
|
shutil.rmtree(temp_parquet_path) |
|
|
|
|
|
for i in range(0, len(data_lines), chunk_size): |
|
|
chunk = data_lines[i:i + chunk_size] |
|
|
rdd_chunk = spark.sparkContext.parallelize(chunk).repartition(8) |
|
|
df_chunk = spark.read.schema(schema).csv(rdd_chunk, header=False) |
|
|
df_chunk.write.mode("append").parquet(temp_parquet_path) |
|
|
|
|
|
return spark.read.parquet(temp_parquet_path) |
|
|
|
|
|
def resample_dataframe(df, track_each=3600): |
|
|
keep_cols = ["Open time", "Open", "High", "Low", "Close", "Number of trades"] |
|
|
df = df.select(keep_cols) |
|
|
window_spec = Window.orderBy("Open time") |
|
|
df = df.withColumn("row_number", row_number().over(window_spec)) |
|
|
df = df.withColumn("group_id", floor((col("row_number") - 1) / track_each)) |
|
|
aggregations = [ |
|
|
first("Open time").alias("Open time"), |
|
|
first("Open").alias("Open"), |
|
|
max("High").alias("High"), |
|
|
min("Low").alias("Low"), |
|
|
last("Close").alias("Close"), |
|
|
sum("Number of trades").alias("Number of trades") |
|
|
] |
|
|
aggregated_df = df.groupBy("group_id").agg(*aggregations) |
|
|
return aggregated_df.select("Open time", "Open", "High", "Low", "Close", "Number of trades") |
|
|
|
|
|
def process_financial_data(bucket_name="minio-ngrok-bucket", file_name="BTCUSDT-1s-2025-09.csv", |
|
|
temp_parquet_path="temp/temp_parquet_chunks", |
|
|
output_parquet_path="temp/aggregated_output"): |
|
|
minio_client = sign_in() |
|
|
spark = initialize_spark_session() |
|
|
|
|
|
try: |
|
|
csv_lines = get_minio_data(minio_client, bucket_name, file_name) |
|
|
print(f"Fetched CSV data from MinIO: {len(csv_lines)} lines") |
|
|
df = create_dataframe_from_csv(spark, csv_lines, temp_parquet_path) |
|
|
print("Created Spark DataFrame from CSV data.") |
|
|
aggregated_df = resample_dataframe(df) |
|
|
print("Resampled DataFrame with OHLC aggregations.") |
|
|
|
|
|
|
|
|
os.makedirs(os.path.dirname(output_parquet_path), exist_ok=True) |
|
|
aggregated_df.write.mode("overwrite").parquet(output_parquet_path) |
|
|
print(f"Saved aggregated DataFrame to {output_parquet_path}") |
|
|
|
|
|
|
|
|
if not os.path.exists(output_parquet_path) or not os.path.isdir(output_parquet_path): |
|
|
raise FileNotFoundError(f"Parquet directory {output_parquet_path} was not created or is not a directory.") |
|
|
else: |
|
|
print(f"Verified: Parquet directory exists at {output_parquet_path}") |
|
|
|
|
|
return output_parquet_path |
|
|
|
|
|
except Exception as e: |
|
|
print(f"Error in process_financial_data: {e}") |
|
|
raise |
|
|
finally: |
|
|
spark.stop() |
|
|
|
|
|
if __name__ == "__main__": |
|
|
|
|
|
output_parquet_path = process_financial_data() |
|
|
print(output_parquet_path) |
