<jupyter_start><jupyter_text>Grundlagen von Sparks Dataframe<jupyter_code>#hide
# Autoreload funktioniert nicht bei XDebug
#%load_ext autoreload
#%autoreload 2
#hide
from nbdev.showdoc import *<jupyter_output><empty_output><jupyter_text>Init Spark Context<jupyter_code>from udemy_spark.spark_core import *
spark = get_spark_session() # Session anlegen
spark # Ausgabe der wichtigsten Session Informationen<jupyter_output><empty_output><jupyter_text>Laden von Dummy Daten<jupyter_code>df_people = spark.read.json('data/people.json')
df_apple  = spark.read.csv('data/appl_stock.csv',inferSchema=True,header=True)<jupyter_output><empty_output><jupyter_text>Informationen von DF abfragen<jupyter_code>df_people.show()        # zeigt die Daten
df_people.printSchema() # zeigt das Schema
df_people.columns       # Attribut mit den Columnnames
df_people.describe()    # Beschreibung<jupyter_output>+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)<jupyter_text>Laden von Daten JSON typisiert lesen<jupyter_code>from pyspark.sql.types import (StructField,StringType,IntegerType,StructType)
data_schema = [StructField("age", IntegerType(), True),StructField("name", StringType(), True)]
final_struc = StructType(fields=data_schema)
df_people = spark.read.json('data/people.json', schema=final_struc)
df_people.printSchema()<jupyter_output>root
 |-- age: integer (nullable = true)
 |-- name: string (nullable = true)<jupyter_text>Inhalt von DF anzeigen<jupyter_code>df_people.head(2) # liefert eine Liste mit Row-Elementen
df_people.show(2) # zeigt 2 Zeilen -> df_people.show() zeigt alles<jupyter_output>+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
+----+-------+
only showing top 2 rows<jupyter_text>Zugriff auf Spalten<jupyter_code>print(df_people['age'])
print(type(df_people['age']))
df_people.select('age')#.show() -> df_people.select -> returns a dataframe, not just a  column
df_people.select(['age','name']) #.show()<jupyter_output><empty_output><jupyter_text>Daten filtern  Like SQL-Where<jupyter_code>df_apple.filter("Close<500") # Die Filter koennen wie Where-Klauseln von SQL Statements definiert werden
df_apple.filter("Close<500").select(['Open','Close']).show(3)  # filter gibt wieder ein DF zurueck.<jupyter_output>+----------+----------+
|      Open|     Close|
+----------+----------+
|213.429998|214.009998|
|214.599998|214.379993|
|214.379993|210.969995|
+----------+----------+
only showing top 3 rows<jupyter_text>Standard Python / Pandas Style<jupyter_code>df_apple.filter(df_apple['Close'] < 500).show(2) # standard Pandas Style
df_apple.filter( (df_apple["Close"] < 200) & (df_apple['Open'] > 200) ).show(2) # Achtung: die Klammern sind wichtig, anstelle von 'and' und 'or' & und | verwenden
df_apple.filter( (df_apple["Close"] < 200) & ~(df_apple['Open'] > 200) ).show(2) # ~ ist not Funktion: hier "nicht groesser als 200"<jupyter_output>+-------------------+----------+----------+------------------+----------+---------+------------------+
|               Date|      Open|      High|               Low|     Close|   Volume|         Adj Close|
+-------------------+----------+----------+------------------+----------+---------+------------------+
|2010-01-04 00:00:00|213.429998|214.499996|212.38000099999996|214.009998|123432400|         27.727039|
|2010-01-05 00:00:00|214.599998|215.589994|        213.249994|214.379993|150476200|27.774976000000002|
+-------------------+----------+----------+------------------+----------+---------+------------------+
only showing top 2 rows

+-------------------+------------------+----------+----------+----------+---------+------------------+
|               Date|              Open|      High|       Low|     Close|   Volume|         Adj Close|
+-------------------+------------------+----------+----------+----------+---------+------------------+
|2010-01-22 00:00:00|206.78000600000001|207.4999[...]<jupyter_text>Rueckgabe als Python Objekte<jupyter_code>result = df_apple.filter(df_apple["Low"] == 197.16).collect()
print(type(result)) # Python Liste
print(type(result[0])) # mit Spark Row Objekten
print(result[0].asDict()) # eine Row kann direkt als Dict zurueckgegeben werden
for item in result[0]: # ueber die Spalten in einer Row kann auch direkt iteriert werden
    print(item, end=' | ')<jupyter_output>{'Date': datetime.datetime(2010, 1, 22, 0, 0), 'Open': 206.78000600000001, 'High': 207.499996, 'Low': 197.16, 'Close': 197.75, 'Volume': 220441900, 'Adj Close': 25.620401}
2010-01-22 00:00:00 | 206.78000600000001 | 207.499996 | 197.16 | 197.75 | 220441900 | 25.620401 |<jupyter_text>Manipulation von DF  Neue Kolonnen erzeugen<jupyter_code>df_people.withColumn('newage', df_people['age']).show()   # erzeugt ein neues DF mit einer neue Spalte basierend auf 'age'
df_people.withColumn('newage', df_people['age']*2).show()   # erzeugt ein neues DF mit einer neue Spalte basierend auf 'age'
df_people.withColumnRenamed('age','newage').show() # erzeugt neues DF mit umbenannter Spalte<jupyter_output><empty_output><jupyter_text>SQL verwenden<jupyter_code># Registriere das Datenframe als voruebergehende SQL-Ansicht
df_people.createOrReplaceTempView("people")
sql_results = spark.sql("SELECT * FROM people") # sql_results ist ein Dataframe
sql_results.show()<jupyter_output>+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+
{'fraction_non_alphanumeric': 0.24532413292173597, 'fraction_numerical': 0.07735609224623206, 'mean_word_length': 3.0292611558156546, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 5, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '93298', 'license': 'Apache-2.0', 'n_tokens_mistral': 2165, 'n_tokens_neox': 1867, 'n_words': 364, 'path': '01_df_basics_dbg.ipynb', 'repo_name': 'HansjoergW/udemy_spark', 'size': '5507'}