<jupyter_start><jupyter_text>Import Statement<jupyter_code>from pyspark.sql import SQLContext
from handyspark import *
from pyspark.sql import functions as sf
from matplotlib import pyplot as plt
from pyspark.sql.functions import col, avg
import pandas as pd
from pyspark.sql.window import Window
log4jLogger = sc._jvm.org.apache.log4j
LOGGER = log4jLogger.LogManager.getLogger(__name__)
LOGGER.error("pyspark script logger initialized")
base_path = "/home/darkmatter/Desktop/smart-meters-in-london/"
sqlcontext = SQLContext(sc)
household_info = sqlcontext.read.csv(base_path+"informations_households.csv",header=True,inferSchema=True)
household_mini = sc.parallelize(household_info.take(15)).toDF()
# household_mini = household_info
household_mini.show()
column_list = []
for i in range(48):
    column_list.append("hh_"+str(i))
column_list
new_column_list = []
for i in range(1,49):
    if i<20:
        new_column_list.append("0"+str(i*0.5))
    else:
        new_column_list.append(str(i*0.5))
flag = 0
avg_house_data = []
df_file = household_mini.select("file").distinct()
# exprs = {x: "avg" for x in new_column_list}
exprs1 = [avg(x) for x in column_list[0:40]]
exprs2 = [avg(x) for x in column_list[40:48]]        #due to the fact large number of column giving error so divide
for row in df_file.rdd.collect():
    file = row.file
    print(file)
    file_path = base_path + "hhblock_dataset/"+ file+".csv"
    half_hourly_consumption_data = sqlcontext.read.csv(file_path,header=True,inferSchema=True)
    half_hourly_consumption_data.dropna(how='any')
    half_hourly_consumption_data2 = half_hourly_consumption_data.groupBy('LCLid').agg(*exprs2)
    half_hourly_consumption_data = half_hourly_consumption_data.groupBy('LCLid').agg(*exprs1)
    half_hourly_consumption_data = half_hourly_consumption_data.join(half_hourly_consumption_data2,["LCLid"])
    half_hourly_consumption_data.dropna(how='any')
    half_hourly_consumption_data.printSchema()
    if flag == 0:
        avg_house_data = sqlcontext.createDataFrame([],half_hourly_consumption_data.schema)
        flag = 1
    avg_house_data = avg_house_data.union(half_hourly_consumption_data)
half_hourly_consumption_data.show()
flag = 0
avg_house_data = []
block_read = set([])
for row in household_mini.rdd.collect():
    house_id = row.LCLid
    file = row.file
    print(house_id,file)
    file_path = base_path + "hhblock_dataset/"+ file+".csv"
    if file not in block_read:
#         print("hi")
        block_read.add(file)
        half_hourly_consumption_data = sqlContext.read.csv(file_path,header=True,inferSchema=True)
        half_hourly_consumption_data.dropna(how='any')
        for c,n in zip(column_list,new_column_list):
            half_hourly_consumption_data=half_hourly_consumption_data.withColumnRenamed(c,n)
    indiv_house_data = half_hourly_consumption_data.where(col("LCLid") == house_id)
    indiv_house_data = indiv_house_data.toHandy()
    if indiv_house_data.rdd.isEmpty():
        print("Missing Id = {} in file = {}".format(house_id,file))
        continue
    indiv_house_data = sqlcontext.createDataFrame(indiv_house_data.stratify(['LCLid']).cols[new_column_list].mean().reset_index())
#     indiv_house_data.printSchema()
    if flag == 0:
        avg_house_data = sqlcontext.createDataFrame([],indiv_house_data.schema)
        flag = 1
    avg_house_data = avg_house_data.union(indiv_house_data)
for c,n in zip(avg_house_data.columns[1:],new_column_list):
    avg_house_data=avg_house_data.withColumnRenamed(c,n)
avg_house_data.show()
avg_house_data = avg_house_data.toPandas().set_index("LCLid")
# pd.options.display.max_columns = None
avg_house_data.head(15)
avg_house_data.T.plot(figsize=(13,8), legend=False, color='blue',alpha=0.5)
avg_house_data.to_csv(base_path+"avg.csv")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11548556430446194, 'fraction_numerical': 0.011548556430446194, 'mean_word_length': 5.405042016806723, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 5, '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': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '506701', 'license': 'Apache-2.0', 'n_tokens_mistral': 1402, 'n_tokens_neox': 1370, 'n_words': 210, 'path': 'Other_Scrap_file/aa/Average_load_profile.ipynb', 'repo_name': 'Ayush-IITP/London_Smart_Meter_Load_Forecasting', 'size': '3810'}