SparkSession

Vstupním bodem pro programování Sparku pomocí rozhraní API datové sady a datového rámce SparkSession se dá použít k vytváření datových rámců, registraci datových rámců jako tabulek, spouštění SQL přes tabulky, tabulky mezipaměti a čtení souborů Parquet.

Syntaxe

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

Vlastnosti

Vlastnictví	Description
version	Verze Sparku, na které je tato aplikace spuštěná.
conf	Rozhraní konfigurace modulu runtime pro Spark
catalog	Rozhraní, pomocí kterého může uživatel vytvářet, odstraňovat, měnit nebo dotazovat podkladové databáze, tabulky, funkce atd.
udf	Vrátí UDFRegistration pro registraci UDF.
udtf	Vrátí UDTFRegistration pro registraci UDTF.
dataSource	Vrátí dataSourceRegistration pro registraci zdroje dat.
profile	Vrátí profil pro profilaci výkonu a paměti.
sparkContext	Vrátí základní SparkContext. Pouze klasický režim.
read	Vrátí DataFrameReader, který lze použít ke čtení dat jako datového rámce.
readStream	Vrátí Třídy DataStreamReader, které lze použít ke čtení datových proudů jako streamovaného datového rámce.
streams	Vrátí StreamQueryManager, který umožňuje spravovat všechny aktivní dotazy streamování.
tvf	Vrátí TableValuedFunction pro volání funkcí s hodnotami tabulky (TVFs).

Methods

Metoda	Description
createDataFrame(data, schema, samplingRatio, verifySchema)	Vytvoří datový rámec ze sady RDD, seznamu, datového rámce pandas, numpy ndarray nebo pyarrow Table.
sql(sqlQuery, args, **kwargs)	Vrátí datový rámec představující výsledek daného dotazu.
table(tableName)	Vrátí zadanou tabulku jako datový rámec.
range(start, end, step, numPartitions)	Vytvoří datový rámec s jedním sloupcem LongType s názvem id, který obsahuje prvky v oblasti.
newSession()	Vrátí novou SparkSession s samostatným SQLConf, registrovanými dočasnými zobrazeními a funkcemi definovanými uživatelem, ale sdílenou mezipamětí SparkContext a tabulek. Pouze klasický režim.
getActiveSession()	Vrátí aktivní SparkSession pro aktuální vlákno.
active()	Vrátí aktivní nebo výchozí SparkSession pro aktuální vlákno.
stop()	Zastaví základní SparkContext.
addArtifacts(*path, pyfile, archive, file)	Přidá artefakty do relace klienta.
interruptAll()	Přeruší všechny operace této relace, které jsou aktuálně spuštěné na serveru.
interruptTag(tag)	Přeruší všechny operace této relace danou značkou.
interruptOperation(op_id)	Přeruší operaci této relace s daným ID operace.
addTag(tag)	Přidá značku, která se přiřadí ke všem operacím spuštěným tímto vláknem v této relaci.
removeTag(tag)	Odebere značku dříve přidanou pro operace spuštěné tímto vláknem.
getTags()	Získá značky, které jsou aktuálně nastaveny tak, aby byly přiřazeny ke všem operacím spuštěným tímto vláknem.
clearTags()	Vymaže značky operací aktuálního vlákna.

Budovatel

Metoda	Description
config(key, value)	Nastaví možnost konfigurace. Možnosti se automaticky rozšíří do vlastní konfigurace SparkConf i SparkSession.
master(master)	Nastaví hlavní adresu URL Sparku pro připojení.
remote(url)	Nastaví vzdálenou adresu URL Sparku pro připojení přes Spark Connect.
appName(name)	Nastaví název aplikace, která se zobrazí ve webovém uživatelském rozhraní Sparku.
enableHiveSupport()	Umožňuje podporu Hivu, včetně připojení k trvalému metastoru Hive.
getOrCreate()	Získá existující SparkSession nebo, pokud neexistuje žádný, vytvoří nový na základě možností nastavených v tomto tvůrci.
create()	Vytvoří novou SparkSession.

Příklady

spark = (
    SparkSession.builder
        .master("local")
        .appName("Word Count")
        .config("spark.some.config.option", "some-value")
        .getOrCreate()
)

spark.sql("SELECT * FROM range(10) where id > 7").show()

+---+
| id|
+---+
|  8|
|  9|
+---+

spark.createDataFrame([('Alice', 1)], ['name', 'age']).show()

+-----+---+
| name|age|
+-----+---+
|Alice|  1|
+-----+---+

spark.range(1, 7, 2).show()

+---+
| id|
+---+
|  1|
|  3|
|  5|
+---+