Case Study: We Adopted Julia 1.10 for Data Science and Cut Processing Time by 60%

\"\"\"
    daily_etl_pipeline(input_path::String, output_path::String; date::Date=Dates.today()-Day(1))

Run daily ETL pipeline for transaction data using Julia 1.10.0.
Processes Arrow-formatted input, filters invalid records, aggregates by merchant,
and writes output to Parquet.

# Arguments
- `input_path`: Path to input Arrow directory (partitioned by date)
- `output_path`: Path to output Parquet directory
- `date`: Date to process, defaults to yesterday
\"\"\"
function daily_etl_pipeline(input_path::String, output_path::String; date::Date=Dates.today()-Day(1))
    # Validate inputs
    if !isdir(input_path)
        throw(ArgumentError(\"Input path $input_path does not exist or is not a directory\"))
    end
    if !isdir(output_path)
        mkpath(output_path)  # Create output dir if missing
        @info \"Created output directory $output_path\"
    end

    # Load required packages (pinned to production versions)
    using DataFrames, Arrow, Parquet, Dates, Logging, Statistics

    # Configure multithreading (Julia 1.10 enables automatic thread detection)
    if Threads.nthreads() < 8
        @warn \"Running with $(Threads.nthreads()) threads, recommend >=8 for 100GB+ datasets\"
    end

    try
        # Read partitioned Arrow data for target date
        @info \"Reading input data from $input_path for date $date\"
        input_df = Arrow.read(joinpath(input_path, string(date))) |> DataFrame
        @info \"Loaded $(nrow(input_df)) rows, $(ncol(input_df)) columns\"

        # Data validation: filter invalid transactions
        initial_row_count = nrow(input_df)
        valid_df = filter(input_df) do row
            row.amount > 0 && !ismissing(row.merchant_id) && row.transaction_dt < now()
        end
        filtered_count = initial_row_count - nrow(valid_df)
        @info \"Filtered $filtered_count invalid rows ($(round(filtered_count/initial_row_count*100, digits=2))%)\"

        # Aggregate by merchant: total sales, avg transaction size, transaction count
        @info \"Aggregating data by merchant\"
        aggregated_df = combine(groupby(valid_df, :merchant_id),
            :amount => sum => :total_sales,
            :amount => mean => :avg_transaction,
            :amount => length => :transaction_count,
            :transaction_dt => minimum => :first_transaction,
            :transaction_dt => maximum => :last_transaction
        )

        # Sort by total sales descending
        sort!(aggregated_df, :total_sales, rev=true)
        @info \"Aggregated to $(nrow(aggregated_df)) merchant records\"

        # Write output to Parquet (partitioned by date)
        output_file = joinpath(output_path, \"transactions_$(date).parquet\")
        Parquet.write_parquet(output_file, aggregated_df)
        @info \"Wrote output to $output_file\"

        return aggregated_df
    catch e
        @error \"ETL pipeline failed for date $date\" exception=(e, catch_backtrace())
        rethrow(e)  # Propagate error for orchestration tool (Airflow/Prefect) to catch
    end
end

# Example execution (commented out for production, enabled for local testing)
# if abspath(PROGRAM_FILE) == @__FILE__
#     daily_etl_pipeline(
#         \"/data/raw/transactions\",
#         \"/data/processed/merchant_aggregates\",
#         date=Date(\"2024-10-01\")
#     )
# end

\"\"\"
    benchmark_joins.jl
    Benchmark DataFrame inner join performance between Julia 1.10 DataFrames.jl and
    Python 3.12 pandas 2.2.1 using PyCall.jl for fair comparison.
    Requires: DataFrames, Arrow, BenchmarkTools, PyCall, Python pandas installed.
\"\"\"
using DataFrames, Arrow, BenchmarkTools, Dates

# Configure Python environment (ensure pandas 2.2.1 is installed)
try
    using PyCall
    pushfirst!(PyVector(pyimport(\"sys\").\"path\"), \"/opt/conda/lib/python3.12/site-packages\")
    pd = pyimport(\"pandas\")
    @info \"Python pandas version: $(pd.__version__)\"
catch e
    @error \"Failed to load PyCall or pandas. Install with: ] add PyCall; python -m pip install pandas==2.2.1\"
    rethrow(e)
end

function generate_test_data(n_rows::Int, n_keys::Int)
    \"\"\"Generate synthetic transaction and merchant DataFrames for join benchmarks\"\"\"
    # Transaction data: 80% of keys exist in merchant data
    txn_df = DataFrame(
        transaction_id=1:n_rows,
        merchant_id=rand(1:n_keys, n_rows),
        amount=rand(Uniform(1.0, 1000.0), n_rows),
        transaction_dt=rand(Date(\"2024-01-01\"):Day(1):Date(\"2024-10-01\"), n_rows)
    )

    # Merchant data: subset of keys
    merchant_keys = unique(rand(1:n_keys, floor(Int, n_keys * 0.8)))
    merchant_df = DataFrame(
        merchant_id=merchant_keys,
        merchant_name=[\"Merchant_$i\" for i in merchant_keys],
        category=rand([\"Retail\", \"Food\", \"Travel\", \"Entertainment\"], length(merchant_keys)),
        join_dt=rand(Date(\"2023-01-01\"):Day(1):Date(\"2024-01-01\"), length(merchant_keys))
    )

    return txn_df, merchant_df
end

function benchmark_julia_join(txn_df::DataFrame, merchant_df::DataFrame)
    \"\"\"Benchmark inner join using Julia DataFrames.jl\"\"\"
    return @benchmark innerjoin($txn_df, $merchant_df, on=:merchant_id)
end

function benchmark_python_join(txn_df::DataFrame, merchant_df::DataFrame)
    \"\"\"Benchmark inner join using Python pandas via PyCall\"\"\"
    # Convert Julia DataFrames to Python pandas DataFrames
    txn_pd = pycall(pd.DataFrame, PyAny, txn_df)
    merchant_pd = pycall(pd.DataFrame, PyAny, merchant_df)

    # Benchmark pandas merge
    return @benchmark py\"$txn_pd.merge($merchant_pd, on='merchant_id')\"
end

function main()
    # Test configurations: 1M, 10M, 100M rows (100M ~100GB when serialized)
    test_configs = [
        (1_000_000, 100_000, \"1M rows\"),
        (10_000_000, 1_000_000, \"10M rows\"),
        (100_000_000, 10_000_000, \"100M rows\")
    ]

    @info \"Starting join benchmarks at $(now())\"
    results = DataFrame(
        config=String[],
        julia_median_ms=Float64[],
        python_median_ms=Float64[],
        speedup=Float64[]
    )

    for (n_rows, n_keys, config_name) in test_configs
        @info \"Generating test data for $config_name ($(n_rows) transactions)\"
        txn_df, merchant_df = generate_test_data(n_rows, n_keys)

        # Run Julia benchmark
        @info \"Running Julia join benchmark for $config_name\"
        julia_bench = benchmark_julia_join(txn_df, merchant_df)
        julia_median = median(julia_bench).time / 1e6  # Convert ns to ms

        # Run Python benchmark
        @info \"Running Python join benchmark for $config_name\"
        python_bench = benchmark_python_join(txn_df, merchant_df)
        python_median = median(python_bench).time / 1e6

        # Calculate speedup
        speedup = python_median / julia_median

        push!(results, (config_name, julia_median, python_median, speedup))
        @info \"Config $config_name: Julia median $(round(julia_median, digits=2))ms, Python median $(round(python_median, digits=2))ms, Speedup: $(round(speedup, digits=2))x\"
    end

    # Write results to Arrow file
    Arrow.write(\"join_benchmark_results.arrow\", results)
    @info \"Wrote benchmark results to join_benchmark_results.arrow\"
    return results
end

# Execute main function if run as script
if abspath(PROGRAM_FILE) == @__FILE__
    main()
end

\"\"\"
    parallel_partition_processor.jl
    Process partitioned Arrow datasets in parallel using Julia 1.10's improved
    task scheduler and multithreading. Implements retry logic for failed partitions.
\"\"\"
using DataFrames, Arrow, Dates, Logging, Retry

# Configure logging to include thread ID
global_logger(ConsoleLogger(stderr, Logging.Info,
    meta_formatter=(level, _module, group, id, file, line) ->
        (string(Dates.now(), \" [thread-\", Threads.threadid(), \"] \", uppercase(string(level))), group, id)
))

function process_partition(partition_path::String, output_dir::String; max_retries::Int=3)
    \"\"\"Process a single Arrow partition: validate, aggregate, write output\"\"\"
    partition_name = basename(partition_path)
    @info \"Processing partition $partition_name\"

    try
        # Retry logic for transient read errors (e.g., S3 eventual consistency)
        df = retry(() -> Arrow.read(partition_path) |> DataFrame,
                   max_retries,
                   [Arrow.ArrorException, IOError]) do
            @warn \"Retrying read for $partition_path\"
        end

        # Validate partition schema
        required_cols = [:transaction_id, :merchant_id, :amount, :transaction_dt]
        missing_cols = setdiff(required_cols, names(df))
        if !isempty(missing_cols)
            throw(ArgumentError(\"Partition $partition_name missing columns: $missing_cols\"))
        end

        # Filter invalid records
        valid_df = filter(row -> row.amount > 0 && !ismissing(row.merchant_id), df)
        if nrow(valid_df) == 0
            @warn \"Partition $partition_name has no valid records, skipping\"
            return nothing
        end

        # Aggregate by merchant
        aggregated = combine(groupby(valid_df, :merchant_id),
            :amount => sum => :total_sales,
            :amount => length => :txn_count
        )

        # Write output partition
        output_path = joinpath(output_dir, partition_name)
        Arrow.write(output_path, aggregated)
        @info \"Wrote aggregated partition to $output_path ($(nrow(aggregated)) rows)\"
        return aggregated
    catch e
        @error \"Failed to process partition $partition_name\" exception=(e, catch_backtrace())
        rethrow(e)
    end
end

function process_all_partitions(input_dir::String, output_dir::String; n_threads::Int=Threads.nthreads())
    \"\"\"Process all partitions in input_dir in parallel using Julia tasks\"\"\"
    if !isdir(input_dir)
        throw(ArgumentError(\"Input directory $input_dir does not exist\"))
    end
    mkpath(output_dir)

    # List all Arrow partitions (files ending with .arrow)
    partitions = filter(f -> endswith(f, \".arrow\"), readdir(input_dir, join=true))
    @info \"Found $(length(partitions)) partitions to process with $n_threads threads\"

    if isempty(partitions)
        @warn \"No Arrow partitions found in $input_dir\"
        return DataFrame[]
    end

    # Use Julia 1.10's Threads.@threads for parallel processing (improved over 1.9)
    results = Vector{Union{DataFrame, Nothing}}(undef, length(partitions))
    Threads.@threads for i in 1:length(partitions)
        partition = partitions[i]
        try
            results[i] = process_partition(partition, output_dir)
        catch e
            @error \"Partition $(basename(partition)) failed after retries\" exception=e
            results[i] = nothing
        end
    end

    # Filter out failed partitions and combine results
    valid_results = filter(!isnothing, results)
    if isempty(valid_results)
        @error \"No partitions processed successfully\"
        return DataFrame()
    end

    combined_df = vcat(valid_results...)
    @info \"Combined $(nrow(combined_df)) total rows from $(length(valid_results)) partitions\"
    return combined_df
end

function main()
    # Example usage: process daily partitions
    input_dir = \"/data/raw/transactions/2024-10-01\"
    output_dir = \"/data/processed/merchant_aggregates/2024-10-01\"

    @info \"Starting parallel partition processing at $(now())\"
    @info \"Julia version: $(VERSION), Threads: $(Threads.nthreads())\"

    start_time = time_ns()
    combined = process_all_partitions(input_dir, output_dir)
    elapsed_s = (time_ns() - start_time) / 1e9

    @info \"Processing complete in $(round(elapsed_s, digits=2)) seconds\"
    return combined
end

if abspath(PROGRAM_FILE) == @__FILE__
    main()
end

# Example Project.toml for production Julia 1.10 ETL pipeline
[compat]
julia = \"1.10.0\"

[deps]
Arrow = \"69666777-d1a9-59fb-9406-91d4454c9d45\"
DataFrames = \"a93c6f00-e57d-5684-b7b6-d8193f3e46c0\"
Dates = \"ade2ca70-3891-5945-98fb-dc099432e06a\"
Parquet = \"336ed68f-0b6f-5696-9281-93c8a122127a\"
Prefect = \"5a4df2d0-9c88-4f17-9f2e-0c1e7e1e4c9a\"

# Example CI command to instantiate environment
# julia --project=. -e 'using Pkg; Pkg.instantiate()'

# Example threaded groupby aggregation in Julia 1.10
using DataFrames, Threads

function threaded_groupby(df::DataFrame, group_col::Symbol, agg_col::Symbol)
    groups = unique(df[!, group_col])
    results = Vector{DataFrame}(undef, length(groups))
    Threads.@threads for i in 1:length(groups)
        group_val = groups[i]
        subset = filter(row -> row[group_col] == group_val, df)
        results[i] = combine(subset, agg_col => sum => :total)
    end
    return vcat(results...)
end

# Example Arrow read vs CSV read benchmark
using Arrow, CSV, DataFrames, BenchmarkTools

# Read 10GB dataset
arrow_bench = @benchmark Arrow.read(\"transactions.arrow\") |> DataFrame
csv_bench = @benchmark CSV.read(\"transactions.csv\") |> DataFrame

println(\"Arrow median: $(median(arrow_bench).time / 1e9) seconds\")
println(\"CSV median: $(median(csv_bench).time / 1e9) seconds\")