# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: PySpark
#     language: python
#     name: pysparkkernel
# ---

# +
# https://www.kaggle.com/bryanb/fifa-player-stats-database?select=FIFA20_official_data.csv

import hsfs

connection = hsfs.connection()
fs = connection.get_feature_store()

# +
from hops import hdfs
from pyspark.sql.functions import lower, col, lit


defaware = ['22', '21', '20', '19', '18', '17']
for year in defaware:
    df=spark.read.format("csv").option("header","true").load(hdfs.project_path()+"Resources/FIFA" + year + "_official_data.csv")
    if year == "22" or year == "21" or year == "20":
        df = df.drop(col("DefensiveAwareness"))
        
    df = df.toDF(*[c.lower() for c in df.columns])
    df = df.toDF(*[c.replace(' ', '_') for c in df.columns])
        
    df.coalesce(1).write.mode("overwrite").csv(hdfs.project_path()+"Resources/FIFA" + year + "_official_data_cleaned.csv",header=True) 


# +
from pyspark.sql.functions import *

# Fix 2017, which does not have a 'release_clause' column. Take the data from 2018
df1 = spark.read.format("csv").option("header","true").load(hdfs.project_path()+"Resources/FIFA17_official_data_cleaned.csv/")
df2 = spark.read.format("csv").option("header","true").load(hdfs.project_path()+"Resources/FIFA18_official_data_cleaned.csv/")

df1 = df1.alias('df1')
df2 = df2.alias('df2')

df1 = df1.join(df2, df1.id == df2.id, "left").select('df1.*', 'df2.release_clause')
df1 = df1.na.fill(value="€1.0M", subset=['release_clause'])

df1.show(n=2, truncate=False, vertical=True)
# -

spark.catalog.clearCache()

df1.coalesce(1).write.mode("overwrite").csv(hdfs.project_path()+"Resources/FIFA17_official_data_cleaned.csv",header=True)

hdfs.rmr(hdfs.project_path()+"Resources/FIFA17_official_data_cleaned.csv")
hdfs.move(hdfs.project_path()+"Resources/FIFA17_official_data_cleaned2.csv", hdfs.project_path()+"Resources/FIFA17_official_data_cleaned.csv")

# +
df17 = spark.read.format("csv").option("header","true").load(hdfs.project_path()+"Resources/FIFA17_official_data_cleaned.csv/")

fg = fs.create_feature_group("fifa",
                        version=1,
                        description="Fifa players",
                        primary_key=['id'],
                        statistics_config=True,
                        online_enabled=True)
fg.save(df17)
# -

years = ['18', '19', '20', '21', '22']
for year in years:
    df=spark.read.format("csv").option("header","true").load(hdfs.project_path()+"Resources/FIFA" + year + "_official_data_cleaned.csv")
    fg.insert(df)



StarcoderdataJupyterScriptsDedupFiltered