<jupyter_start><jupyter_text>Replication of paper "Econometrics at Scale - Spark Up Big Data in Economics", Chapter 4.4 (<NAME> and <NAME>)This notebook implements a distributed time series forecasting exercise in PySpark as described in more detail in the paper "Econometrics at Scale - Spark Up Big Data in Economics" by <NAME> and <NAME>.  Create dictionary with config parameters<jupyter_code>import os

# initialize dictionary with config parameters
config = {}

# Define working directory
wdir = 's3://my-bucket/'

# Define AWS S3 endpoint for your region
config['s3_host'] = 's3.eu-central-1.amazonaws.com'

# Define Path to store results
config['path_training_data_csv'] = os.path.join(wdir, 'data/time_series/rawdata_small.csv')
config['path_training_data_parquet'] = os.path.join(wdir, 'output/time_series/parquet/')
config['path_forecasts'] = os.path.join(wdir, 'output/time_series/forecasts/')
config['path_models'] = os.path.join(wdir, 'output/time_series/ models/')
config['path_code'] = os.path.join(wdir, 'scripts/time_ series/')

# Define series and evaluation lengths
config['len_series'] = 1000
config['len_eval'] = 50<jupyter_output><empty_output><jupyter_text>Install necessary Python packages on all cluster nodes<jupyter_code>sc.install_pypi_package("s3fs")
sc.install_pypi_package("joblib")
sc.install_pypi_package("s3io")
sc.install_pypi_package("pandas")
sc.install_pypi_package("fastparquet")
sc.install_pypi_package("statsmodels")<jupyter_output><empty_output><jupyter_text>Partition and save dataset to Parquet formatIn this step, we load the sample csv file into a Spark DataFrame, repartition the DataFrame by 'ID' column and save the data in Parquet file format<jupyter_code>df = spark.read.csv(config['path_training_data_csv'],header = True, inferSchema=True)

df.repartition("ID").write.option("compression", "gzip").mode("overwrite")\
    .partitionBy("ID").parquet(config['path_training_data_parquet'])<jupyter_output><empty_output><jupyter_text>Add Python module to Spark context for distributed execution<jupyter_code>spark.sparkContext.addPyFile(config['path_code'] + '/fit_model_and_forecast.py')<jupyter_output><empty_output><jupyter_text>Perform model fitting and forecasting (distributed execution scheme)In this step, we run the forecasting exercise in a distributed fashion<jupyter_code># Load time series data into Spark dataframe
df = spark.read.parquet(config['path_training_data_parquet'])

# Create RDD with dictinct IDs and repartition dataframe into 100 chunks
time_series_ids = df.select("ID").distinct().repartition(100).rdd

# Function to import Python module on Spark executor for parallel forecasting
def import_module_on_spark_executor(time_series_ids, config):
    from fit_model_and_forecast import fit_model_and_forecast
    return fit_model_and_forecast(time_series_ids, config, cloud=True)

# Parallel model fitting and forecasting
time_series_ids.foreach(lambda x: import_module_on_spark_executor(x, config))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10326449033977349, 'fraction_numerical': 0.0069953364423717525, 'mean_word_length': 7.679190751445087, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '374387', 'license': 'MIT', 'n_tokens_mistral': 974, 'n_tokens_neox': 935, 'n_words': 235, 'path': 'chapter_4.4/pyspark_demo_4.4.ipynb', 'repo_name': 'benjaminbluhm/econometrics_at_scale', 'size': '3022'}