<jupyter_start><jupyter_text>Investor - Flow of Funds - US  Introduction:Special thanks to: https://github.com/rgrp for sharing the dataset. Step 1. Import the necessary libraries<jupyter_code>import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('funds').getOrCreate()
spark<jupyter_output><empty_output><jupyter_text>Step 2. Import the dataset from this [address](https://raw.githubusercontent.com/datasets/investor-flow-of-funds-us/master/data/weekly.csv).<jupyter_code>from pyspark import SparkFiles<jupyter_output><empty_output><jupyter_text>Step 3. Assign it to a variable called<jupyter_code>url = "https://raw.githubusercontent.com/datasets/investor-flow-of-funds-us/master/data/weekly.csv"

spark.sparkContext.addFile(url)

df = spark.read.csv(SparkFiles.get("weekly.csv"), header=True, inferSchema=True, sep=',')
df.show(5)<jupyter_output>+----------+------------+---------------+------------+------+----------+------------+--------------+-----+
|      Date|Total Equity|Domestic Equity|World Equity|Hybrid|Total Bond|Taxable Bond|Municipal Bond|Total|
+----------+------------+---------------+------------+------+----------+------------+--------------+-----+
|2012-12-05|       -7426|          -6060|       -1367|   -74|      5317|        4210|          1107|-2183|
|2012-12-12|       -8783|          -7520|       -1263|   123|      1818|        1598|           219|-6842|
|2012-12-19|       -5496|          -5470|         -26|   -73|       103|        3472|         -3369|-5466|
|2012-12-26|       -4451|          -4076|        -375|   550|      2610|        3333|          -722|-1291|
|2013-01-02|      -11156|          -9622|       -1533|  -158|      2383|        2103|           280|-8931|
+----------+------------+---------------+------------+------+----------+------------+--------------+-----+
only showing top 5 rows<jupyter_text>Step 4.  What is the frequency of the dataset?<jupyter_code>#Weeks<jupyter_output><empty_output><jupyter_text>Step 5. Set the column Date as the index.  Step 6. What is the type of the index?<jupyter_code>df.dtypes<jupyter_output><empty_output><jupyter_text>Step 7. Set the index to a DatetimeIndex type<jupyter_code>import pyspark.sql.types as T
df = df.withColumn('Date', df.Date.cast(T.DateType()))
df.show(5)
df.printSchema()<jupyter_output>root
 |-- Date: date (nullable = true)
 |-- Total Equity: integer (nullable = true)
 |-- Domestic Equity: integer (nullable = true)
 |-- World Equity: integer (nullable = true)
 |-- Hybrid: integer (nullable = true)
 |-- Total Bond: integer (nullable = true)
 |-- Taxable Bond: integer (nullable = true)
 |-- Municipal Bond: integer (nullable = true)
 |-- Total: integer (nullable = true)<jupyter_text>Step 8.  Change the frequency to monthly, sum the values and assign it to monthly.<jupyter_code>import pyspark.sql.functions as F
df_new = df.withColumn('month', F.month('Date'))\
            .withColumn('year', F.year('Date'))
df_new.show(5)
cols = ['year','month']
df_month_sampled = df_new.withColumn('Yr_Mo', F.concat_ws("-",*cols)).drop("Date","year","month")
df_month_sampled.show(5)
monthly = df_month_sampled.groupBy("Yr_Mo").sum().orderBy("Yr_Mo")
monthly.show(5)<jupyter_output>+-------+-----------------+--------------------+-----------------+-----------+---------------+-----------------+-------------------+----------+
|  Yr_Mo|sum(Total Equity)|sum(Domestic Equity)|sum(World Equity)|sum(Hybrid)|sum(Total Bond)|sum(Taxable Bond)|sum(Municipal Bond)|sum(Total)|
+-------+-----------------+--------------------+-----------------+-----------+---------------+-----------------+-------------------+----------+
|2012-12|           -26156|              -23126|            -3031|        526|           9848|            12613|              -2765|    -15782|
| 2013-1|             3661|               -1627|             5288|       2730|          12149|             9414|               2735|     18540|
|2014-11|            -2753|               -7239|             4485|        729|          14528|            11566|               2962|     12502|
| 2014-4|            10842|                1048|             9794|       4931|           8493|             7193|               1300|    [...]<jupyter_text>Step 9. You will notice that it filled the dataFrame with months that don't have any data with NaN. Let's drop these rows.<jupyter_code>monthly = df_month_sampled.dropna().groupBy("Yr_Mo").sum().orderBy("Yr_Mo")
monthly.show(5)<jupyter_output>+-------+-----------------+--------------------+-----------------+-----------+---------------+-----------------+-------------------+----------+
|  Yr_Mo|sum(Total Equity)|sum(Domestic Equity)|sum(World Equity)|sum(Hybrid)|sum(Total Bond)|sum(Taxable Bond)|sum(Municipal Bond)|sum(Total)|
+-------+-----------------+--------------------+-----------------+-----------+---------------+-----------------+-------------------+----------+
|2012-12|           -26156|              -23126|            -3031|        526|           9848|            12613|              -2765|    -15782|
| 2013-1|             3661|               -1627|             5288|       2730|          12149|             9414|               2735|     18540|
|2014-11|            -2753|               -7239|             4485|        729|          14528|            11566|               2962|     12502|
| 2014-4|            10842|                1048|             9794|       4931|           8493|             7193|               1300|    [...]<jupyter_text>Step 10. Good, now we have the monthly data. Now change the frequency to year.<jupyter_code>df_new.columns
cols = ['year','month']
df_year_sampled = df_new.drop("Date","month").dropna().groupBy('year').sum().orderBy('year')
df_year_sampled.show(5)<jupyter_output>+----+-----------------+--------------------+-----------------+-----------+---------------+-----------------+-------------------+----------+---------+
|year|sum(Total Equity)|sum(Domestic Equity)|sum(World Equity)|sum(Hybrid)|sum(Total Bond)|sum(Taxable Bond)|sum(Municipal Bond)|sum(Total)|sum(year)|
+----+-----------------+--------------------+-----------------+-----------+---------------+-----------------+-------------------+----------+---------+
|2012|           -26156|              -23126|            -3031|        526|           9848|            12613|              -2765|    -15782|     8048|
|2013|             3661|               -1627|             5288|       2730|          12149|             9414|               2735|     18540|     4026|
|2014|              330|              -44689|            45021|      19570|          59890|            44994|              14896|     79787|    52364|
|2015|            15049|              -10459|            25508|       7280|          26028|   [...]
{'fraction_non_alphanumeric': 0.2851654846335697, 'fraction_numerical': 0.09958628841607566, 'mean_word_length': 1.3023809523809524, 'pattern_counts': {'":': 0, '<': 32, '<?xml version=': 0, '>': 32, 'https://': 3, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '637908', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 2368, 'n_tokens_neox': 1929, 'n_words': 391, 'path': '09_Time_Series/Investor_Flow_of_Funds_US/PySparkSolution.ipynb', 'repo_name': 'dhruv-agg/pyspark_practice', 'size': '6768'}