<jupyter_start><jupyter_text>Step 2 - QuestionsUsing the global weather data, answer the following:1. Which country had the hottest average mean temperature over the year?2. Which country had the most consecutive days of tornadoes/funnel cloudformations?3. Which country had the second highest average mean wind speed over the year?<jupyter_code>import pyspark.sql.functions as F
data_full.withColumn(
    'month', F.month(F.col('YEARMODA'))
).groupBy(
    'COUNTRY_FULL', 'month'
).agg(
    # possibly multiple weather stations (unverified)
    # thus aggregate by monthly first
    # ideally aggregate by day -> by month -> by year
    # leaving that for later if there's time
    F.mean('TEMP').alias('mean_monthly_TEMP')
).groupBy(
    'COUNTRY_FULL'
).agg(
    F.mean('mean_monthly_TEMP').alias('monthly_TEMP')
).orderBy(
    F.col('monthly_TEMP'), ascending=False
).show()<jupyter_output>+--------------------+-----------------+
|        COUNTRY_FULL|     monthly_TEMP|
+--------------------+-----------------+
|            DJIBOUTI|90.23198015990424|
|                CHAD|87.14639472082757|
|               SUDAN|85.07542338470381|
|               NIGER|85.03855115744923|
| JUAN DE NOVA ISLAND|84.51182777650894|
|                MALI| 84.4698450533835|
|         EL SALVADOR|84.44803416216793|
|              TUVALU|84.34886658865116|
|BRITISH INDIAN OC...| 83.9194399820983|
|             TOKELAU|83.84275418635283|
|      CAYMAN ISLANDS|83.76236167953455|
|        BURKINA FASO|83.75583106326813|
|            MALDIVES|83.67978358497757|
|           SINGAPORE|83.63143542755746|
|ST. VINCENT AND T...|83.54223237327423|
|            CAMBODIA|83.50356529178372|
|    MARSHALL ISLANDS| 83.2364911782702|
|          MICRONESIA|83.23250023814408|
|             SENEGAL|83.13447681132101|
|            KIRIBATI|83.07528320407322|
+--------------------+-----------------+
only showing top[...]<jupyter_text>Q: Which country had the hottest average mean temperature over the year?A: DJIBOUTImakes, sense, it's in a dry part of Africa Which country had the most consecutive days of tornadoes/funnel cloud formations?<jupyter_code>from pyspark.sql.window import Window
from pyspark.sql.types import BooleanType
from pyspark.sql import DataFrame
from typing import List

from etl.pyspark_utils import consecutive_date_diff_by

# do we have daily observations for each country?
# can only compute for consecutive dates
country_date_diff = consecutive_date_diff_by(
    data_full,
    by_cols=['COUNTRY_FULL'],
    date_col='YEARMODA'
)

# filter for countries with inter_date only null or 1
country_single_consecutive = (
    country_date_diff
    .groupBy('COUNTRY_FULL')
    .agg(F.collect_set('inter_date').alias('set_inter_date'))
).withColumn(
    'is_single_consecutive',
    F.udf(lambda s: s == [1], BooleanType())(F.col('set_inter_date'))
)
country_single_consecutive.show()
data_single_consecutive = data_full.join(
    country_single_consecutive.filter(F.col('is_single_consecutive')),
    on='COUNTRY_FULL',
    how='inner'
).withColumn(
    'has_tornado_or_funnel',
    F.udf(lambda s: s[5] == '1', BooleanType())('FRSHTT')
).select('COUNTRY_FULL', 'YEARMODA', 'has_tornado_or_funnel').distinct()
assert data_single_consecutive.count() > 0
assert data_single_consecutive.filter(F.col('has_tornado_or_funnel')).count() > 0
# https://stackoverflow.com/questions/54445961/pyspark-calculate-streak-of-consecutive-observations
# seems like i need a few windows ???

from pyspark.sql.window import Window

w_base = Window.partitionBy('COUNTRY_FULL').orderBy('YEARMODA')
w_flagged = Window.partitionBy('COUNTRY_FULL', 'has_tornado_or_funnel').orderBy('YEARMODA')

dftemp = data_single_consecutive.withColumn(
    'flag_group',
    # when 2 row counts increment, differences increase if there's a break
    F.row_number().over(w_base) - F.row_number().over(w_flagged)
)
dftemp.show()

# i don't really get the following part
w_streak = Window.partitionBy('COUNTRY_FULL', 'has_tornado_or_funnel', 'flag_group').orderBy('YEARMODA')
dftemp = (
    dftemp
    .withColumn(
        'streak_0',
        F.when(F.col('has_tornado_or_funnel'), 0).otherwise(F.row_number().over(w_streak))
    )
    .withColumn(
        'streak_1',
        F.when(~F.col('has_tornado_or_funnel'), 0).otherwise(F.row_number().over(w_streak))
    )
)
dftemp.show()
dftemp.orderBy('streak_1', ascending=False).show()<jupyter_output>+--------------+-------------------+---------------------+----------+--------+--------+
|  COUNTRY_FULL|           YEARMODA|has_tornado_or_funnel|flag_group|streak_0|streak_1|
+--------------+-------------------+---------------------+----------+--------+--------+
|        CANADA|2019-06-26 00:00:00|                 true|       176|       0|       2|
|         ITALY|2019-10-03 00:00:00|                 true|       275|       0|       2|
|         JAPAN|2019-04-02 00:00:00|                 true|        91|       0|       1|
|    BANGLADESH|2019-08-22 00:00:00|                 true|       233|       0|       1|
|         MALTA|2019-01-19 00:00:00|                 true|        18|       0|       1|
|         JAPAN|2019-12-04 00:00:00|                 true|       338|       0|       1|
|         JAPAN|2019-12-03 00:00:00|                 true|       336|       0|       1|
|        JERSEY|2019-06-12 00:00:00|                 true|       162|       0|       1|
|UNITED KINGDOM|2019-06-12 00:00[...]<jupyter_text>Q: Which country had the most consecutive days of tornadoes/funnel cloud formations?A: Seems to be CANADA. Not sure. ----<jupyter_code>data_full.withColumn(
    'month', F.month(F.col('YEARMODA'))
).groupBy(
    'COUNTRY_FULL', 'month'
).agg(
    F.mean('WDSP').alias('mean_monthly_WDSP')
).groupBy(
    'COUNTRY_FULL'
).agg(
    F.mean('mean_monthly_WDSP').alias('monthly_WDSP')
).orderBy(
    F.col('monthly_WDSP'), ascending=False
).show()<jupyter_output>+--------------------+------------------+
|        COUNTRY_FULL|      monthly_WDSP|
+--------------------+------------------+
|FALKLAND ISLANDS ...|17.853323291643207|
|               ARUBA|15.977568466445604|
|       FAROE ISLANDS|15.004640085348584|
|            BARBADOS|14.104861833330643|
|ST. PIERRE AND MI...|13.964271671947186|
|FRENCH SOUTHERN A...|13.721410918455398|
|          CAPE VERDE|13.689624769963283|
|          MAURITANIA|13.049711560945854|
|     TROMELIN ISLAND|12.996336024862229|
|          ST. HELENA| 12.70270694631401|
|          ANTARCTICA|12.317377902614764|
|             SOMALIA|12.270474730728104|
|COCOS (KEELING) I...| 12.04136264134967|
|            GUERNSEY|12.027518638882889|
|        MAN  <NAME>|11.902185717540952|
|          MONTSERRAT|11.813225809104981|
|             ICELAND|11.793342469451671|
|      <NAME>|11.652829126173785|
|           ST. LUCIA|11.580525361820179|
|            SVALBARD|11.489402906171904|
+--------------------+------------[...]
{'fraction_non_alphanumeric': 0.17582098227259518, 'fraction_numerical': 0.12249346120313863, 'mean_word_length': 2.4956830878618588, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 16, 'https://': 1, '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': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '83290', 'license': 'MIT', 'n_tokens_mistral': 2985, 'n_tokens_neox': 2443, 'n_words': 422, 'path': 'test.ipynb', 'repo_name': 'tianle91/paytm-mle-takehome', 'size': '6891'}