<jupyter_start><jupyter_text>Innlesing og behandling av flere parque-datasett i samme prosessesteg med PysparkI denne noten viser vi hvordan man kan lese inn og behandle et ukjent antall inndatasett ved å bruke python collection datatypes (list, dictionary og tuple) og python for loops for å operasjonalisere steg i en tenkt klargjørings eller analyseprosess.     Innhenter verktøy fra bibliotekImport-stegene henter inn bibliotek med kode og funksjoner utviklet eksternt.<jupyter_code>from datetime import datetime
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql import SQLContext
import pyspark.sql.functions as F<jupyter_output><empty_output><jupyter_text>Kjører metoden read.path for å få oversikt over hvilke parquet datasett som er tilrettelagt i tilknytning veilderens lagringsområde i sky.  Oversikt blir lest inn i egen dataframe - df_datasets.Aktuelt lagringsområde blir lagt inn som parameter (string objekt som vi definerer selv) INNDATA_PATH .<jupyter_code>INNDATA_PATH = '/felles/veiledning/pyspark/eksempler/*'
df_datasets = spark.read.path(INNDATA_PATH)<jupyter_output><empty_output><jupyter_text>df_datasets skrives ut i output vindu.<jupyter_code>df_datasets.show(100, False)<jupyter_output><empty_output><jupyter_text>Finner antall tegn i lengste pathUtleder lengden (antall tegn) for lengste path med aggregate function. "Collecter" lengden (antall tegn) på lengste path inn i listeobjekt liste_max_length_path før det deretter blir lagt inn i integer objekt max_lengde_path. max_lengde_path brukes deretter som parameter i paragraf som avleder variabel filename (blir dataframe navn i notebook). Formål med paragraf er unngå hardkoding i utledning av variabelen filename.<jupyter_code>df = df_datasets.withColumn('lengde_path', F.length(df_datasets.path))
liste_max_lengste_path = df.agg({"lengde_path": "max"}).collect()
max_lengde_path = liste_max_lengste_path[0][0]<jupyter_output><empty_output><jupyter_text>Avleder variabel filname fra variabel path<jupyter_code>df_datasets = df_datasets.withColumn('filename',F.substring(df_datasets.path, len(INNDATA_PATH), (max_lengde_path + 1 - len(INNDATA_PATH))))
df_datasets.show(100, False)<jupyter_output><empty_output><jupyter_text>Seleketerer ut rader bestående av de filnavn som vi ønsker å trekke ut for videre bearbeiding. Viser hvordan man kan bruke liste (Python list) for å gjøre en slik seleksjon.<jupyter_code>filenames = ['areal', 'bnp', 'innbyggerantall/2020']
df_datasets = df_datasets.filter(df_datasets.filename.isin(filenames))
df_datasets.show(100, False)<jupyter_output><empty_output><jupyter_text>Oppretter dictionaryet dict_df og leser så de tre parquet datasettene i output vindu over inn som dataframes i dictionaryet. Filnavnene blir keys i dictionaryet. Avslutningsvis i pargrafen skrives key-navnene ut i output .<jupyter_code>dict_df = {}
for row in df_datasets.rdd.collect():
    dict_df[row.filename] = spark.read.path(row.path)
for key in dict_df:
    print(key)<jupyter_output><empty_output><jupyter_text>Skriver ut innhold i dataframene i dictionaryet  Areal<jupyter_code>dict_df['areal'].show()<jupyter_output><empty_output><jupyter_text>BNP<jupyter_code>dict_df['bnp'].show()<jupyter_output><empty_output><jupyter_text>Innbyggerantall 2020<jupyter_code>dict_df['innbyggerantall/2020'].show()<jupyter_output><empty_output><jupyter_text>Fjerner duplikater og kolonner vi ikke ønsker å ha med videreVi ønsker å sette sammen informasjon fra datasettene i dictinaryet til et nytt datasett. Dette vil vi gjøre ved å koble de sammen via Landkode variabel. Vi fjerner eventuelle dublikater på Landkode og dropper variable som vi ikke trenger/ønsker å ha med på det nye datasettet i innledende steg under. Skriver ut duplikatene i output før vi deretter fjerner de fra dictionaryet Areal<jupyter_code>dict_df['areal'].exceptAll(dict_df['areal'].dropDuplicates(['Land'])).show()
dict_df['areal'] = dict_df['areal'].dropDuplicates(['Landkode']).drop('kilde')<jupyter_output><empty_output><jupyter_text>BNP (Legger inn variable som skal droppes fra dataframe i pyhton tuple)<jupyter_code>dict_df['bnp'].exceptAll(dict_df['bnp'].dropDuplicates(['Land'])).show()
liste_drop_kolonner = ('Land', 'Kilde', 'År')
dict_df['bnp'] = dict_df['bnp'].drop_duplicates(['Landkode']).drop(*liste_drop_kolonner)<jupyter_output><empty_output><jupyter_text>Innbyggerantall (Legger inn variable som skal droppes fra datafrma i python tuple)<jupyter_code>dict_df['innbyggerantall/2020'].exceptAll(dict_df['innbyggerantall/2020'].dropDuplicates(['Land'])).show()
liste_drop_kolonner = ['Land', 'Kilde', 'År']
dict_df['innbyggerantall/2020'] = dict_df['innbyggerantall/2020'].drop_duplicates(['Landkode']).drop(*liste_drop_kolonner)<jupyter_output><empty_output><jupyter_text>Kobler sammen datasettene og etablerer en dataframe vi skal skrive til dapla for senere å utføre en statistisk analyse på.  Kobler først areal og bnp og oppretter dataframen df_areal_bnp. Vi legger ikke denne dataframen inn i dictionaryet. Det hadde imidlertid enkelt latt seg gjøre ved å skrive f.eks. å bytte ut df_areal_bnp med dict['areal_bnp].<jupyter_code>df_areal_bnp = dict_df['areal'].join(dict_df['bnp'], 'Landkode', how='inner')
df_areal_bnp.show()<jupyter_output><empty_output><jupyter_text>Kobler deretter df_areal_bnp med innbyggerantall/2020 dataframen (i dictionaryet). Dermed får vi produsert den dataframen som vi ønsker å utføre en nærmere statistisk analyse på.<jupyter_code>df_sammensatt = df_areal_bnp.join(dict_df['innbyggerantall/2020'], 'Landkode', how='inner')<jupyter_output><empty_output><jupyter_text>Skriver ut innhold i df_sammensatt i output<jupyter_code>df_sammensatt.show()<jupyter_output><empty_output><jupyter_text>Lagrer dataframe som parquet datasett på DaplaSkriver pyspark dataframe til GCS bucket i parquet format. Skriver til hjemmekatalog under /user.<jupyter_code>df_sammensatt.write\
    .option("valuation", "INTERNAL")\
    .option("state", "INPUT")\
    .path('/felles/veiledning/datasett/df_sammensatt_pyspark')


ds = spark.read.path('/felles/veiledning/datasett/*')
ds.show(10, False)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09433962264150944, 'fraction_numerical': 0.008132726089785295, 'mean_word_length': 7.873015873015873, 'pattern_counts': {'":': 1, '<': 69, '<?xml version=': 0, '>': 69, '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': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '21830', 'license': 'CC0-1.0', 'n_tokens_mistral': 2348, 'n_tokens_neox': 2241, 'n_words': 524, 'path': 'utforske/innlesing_og_behandling_av_datasett_pyspark.ipynb', 'repo_name': 'mariuslars/kurs-ssb-dapla-veileder', 'size': '6148'}