Is there an efficient way to join many (1000+) DF?

user_13952204 0
Viewed 95

I'm fairly new to Spark and I have a DF that stacks 8000 time series on top of each other. These time series make more sense as individual columns so I decided to filter them one by one and join them on the timestamp in a new master DF.

So far so good, however as soon as I try that I get a OOM error.

# My stacked DF and the name of the columns
stocks_df.cache()
tickers = ["TS1", "TS2", "TS3", ...]

def filter_for_symbol(ticker):
    security = stocks_df.filter(stocks_df["Symbol"] == ticker).drop("Symbol")
    # Avoids conflict in column names...
    # for f in features:
    #     security = security.withColumnRenamed(f, ticker + "_" + f)
    return security

# Aggregation Loop
acc_df = filter_for_symbol(tickers[0])
for ticker in tickers[1:]:
    security = filter_for_symbol(ticker)
    acc_df = acc_df.join(security, on="Date", how="outer")

What would be a better way of doing what I'm doing in terms of speed and/or memory management?

UPDATE

What I have, stocks_df: dim = (8000 time series * 3125 entries) x 9 features

+-------------+--------+------+-----+------+
|    Date     | Symbol | Open | ... | Close|
+-------------+--------+------+-----+------+
| 2020/01/01  |  TS1   | 10.0 | ... | 12.2 |
| 2020/01/02  |  TS1   | 12.2 | ... | 13.2 |
|    ...      |  ...   | ...  | ... | ...  |
+-------------+--------+------+-----+------+
| 2020/01/01  |  TS2   | 29.9 | ... | 32.2 |
| 2020/01/02  |  TS2   | 32.2 | ... | 19.2 |
|    ...      |  ...   | ...  | ... | ...  |
+-------------+--------+------+-----+------+
|    ...      |  ...   | ...  | ... | ...  |
+-------------+--------+------+-----+------+

What I would like to get: dim = 3125 x (8000 * 9)

+-------------+----------+-------+-----------+-----------+--------+
|    Date     | TS1_Open :  ...  : TS1_Close | TS2_Close |  TS2_..|
+-------------+----------+-------+-----------+-----------+--------+
| 2020/01/01  |   10.0   :  ...  :    12.2   |    12.2   :  ...   |
| 2020/01/02  |   12.2   :  ...  :    13.2   |    13.2   :  ...   |
|    ...      |   ...    :  ...  :    ...    |    ...    :  ...   |
+-------------+----------+-------+-----------+-----------+--------+

What I get instead: java.lang.OOM

pyspark
2 Answers

Assuming there are no duplicate rows in Date + Symbol', you can set them to be the index, then move Symbolto columns. You will get a MultiIndex as columns, containing[('TS1, 'Open'), ('TS1', 'High'), ..., ('TSn', 'Close')]`

df = (
    df.set_index(['Date', 'Symbol'])
    .unstack() # moves the inner-most level by default
)

WARNING

Ifa symbol is not present in some date, unstack will fill that spot in with np.nan. If this happens often you will use more memory than the data really needs. Should be fine in your case but look out when you increase your data size.

user_66928980

What about just pivot your dataframe?

df.show(10, False)

+----------+-----+----+-----+
|Date      |ymbol|Open|Close|
+----------+-----+----+-----+
|2020/01/01|TS1  |10.0|12.2 |
|2020/01/02|TS1  |12.2|13.2 |
|2020/01/01|TS2  |29.9|32.2 |
|2020/01/02|TS2  |32.2|19.2 |
+----------+-----+----+-----+

cols = df.columns[2:]
tickers = ['TS1', 'TS2']

import pyspark.sql.functions as f

df.groupBy('Date') \
  .pivot('ymbol', tickers) \
  .agg(*map(lambda x: f.sum(f.col(x)).alias(x), cols)) \
  .show(10, False)

+----------+--------+---------+--------+---------+
|Date      |TS1_Open|TS1_Close|TS2_Open|TS2_Close|
+----------+--------+---------+--------+---------+
|2020/01/01|10.0    |12.2     |29.9    |32.2     |
|2020/01/02|12.2    |13.2     |32.2    |19.2     |
+----------+--------+---------+--------+---------+
user_118415710
Related