SparkSession: точка входа, конфигурация и master URL

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("MyApp")
    .config("spark.executor.memory", "4g")
    .config("spark.sql.shuffle.partitions", "200")
    .enableHiveSupport()   # HiveCatalog + HiveQL
    .getOrCreate()
)

# Локально для разработки
spark = SparkSession.builder.master("local[*]").getOrCreate()

# Standalone кластер
spark = SparkSession.builder.master("spark://master:7077").getOrCreate()

# YARN - master URL не нужен, берётся из HADOOP_CONF_DIR
spark = SparkSession.builder.master("yarn").getOrCreate()

# Kubernetes
spark = SparkSession.builder \
    .master("k8s://https://k8s-api:6443") \
    .config("spark.kubernetes.container.image", "my-spark:3.5") \
    .getOrCreate()

spark = SparkSession.builder \
    .config("spark.hadoop.fs.s3a.endpoint", "http://minio:9000") \
    .config("spark.hadoop.fs.s3a.access.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.secret.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .getOrCreate()

# Чтение
df = spark.read.parquet("s3a://bucket/data/")
df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = spark.read.json("data.json")
df = spark.read.format("iceberg").load("catalog.db.table")

# Запись
df.write.mode("overwrite").parquet("s3a://bucket/output/")
df.write.mode("append").partitionBy("date").parquet("s3a://bucket/partitioned/")

# Создать временную вьюшку (живёт в рамках SparkSession)
df.createOrReplaceTempView("events")

# Глобальная вьюшка (видна в других SparkSession)
df.createOrReplaceGlobalTempView("global_events")

# SQL-запрос
result = spark.sql("""
    SELECT user_id, COUNT(*) as cnt
    FROM events
    WHERE date >= '2024-01-01'
    GROUP BY user_id
""")

# Читать из глобальной вьюшки
result2 = spark.sql("SELECT * FROM global_temp.global_events")

# Список баз данных
spark.catalog.listDatabases()

# Список таблиц
spark.catalog.listTables("my_database")

# Проверка существования таблицы
spark.catalog.tableExists("my_database.my_table")

# Очистка кэша
spark.catalog.clearCache()
spark.catalog.uncacheTable("events")

# Смена базы данных
spark.catalog.setCurrentDatabase("my_database")

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# Регистрация через SparkSession (доступно в SQL)
def clean_phone(phone):
    return phone.replace("-", "").replace(" ", "") if phone else None

spark.udf.register("clean_phone", clean_phone, StringType())

# Использование в SQL
result = spark.sql("SELECT clean_phone(phone_number) FROM customers")

# Использование в DataFrame API
clean_phone_udf = udf(clean_phone, StringType())
df.withColumn("clean_phone", clean_phone_udf("phone_number"))

# Основная сессия
spark1 = SparkSession.builder.master("local[*]").appName("App").getOrCreate()

# Вторая сессия (отдельный SessionState, общий SparkContext)
spark2 = spark1.newSession()

# Разные конфигурации на уровне сессии
spark1.conf.set("spark.sql.shuffle.partitions", "200")
spark2.conf.set("spark.sql.shuffle.partitions", "50")

# Разные temp views
spark1.createDataFrame([(1, "a")], ["id", "val"]).createTempView("t1")
spark2.sql("SHOW TABLES").show()  # t1 не видна в spark2

# Общий кэш через SharedState
df = spark1.read.parquet("data.parquet").cache()
# spark2 тоже видит кэш - CacheManager общий

spark = SparkSession.builder.getOrCreate()

try:
    # работа с данными
    df = spark.read.parquet("data/")
    df.write.parquet("output/")
finally:
    # Корректная остановка:
    # 1. Завершает все running jobs
    # 2. Останавливает Executor-ы (через SchedulerBackend)
    # 3. Останавливает SparkUI
    # 4. Сохраняет Event Log для History Server
    spark.stop()

import os
from pyspark.sql import SparkSession


def create_spark_session(
    app_name: str,
    env: str = "local",
) -> SparkSession:
    builder = SparkSession.builder.appName(app_name)

    if env == "local":
        builder = (
            builder
            .master("local[*]")
            .config("spark.sql.shuffle.partitions", "4")
            .config("spark.driver.memory", "2g")
        )
    elif env == "dev":
        builder = (
            builder
            .master("yarn")
            .config("spark.executor.instances", "2")
            .config("spark.executor.memory", "4g")
            .config("spark.executor.cores", "2")
            .config("spark.sql.shuffle.partitions", "50")
        )
    elif env == "prod":
        builder = (
            builder
            .master("yarn")
            .config("spark.executor.instances", "20")
            .config("spark.executor.memory", "16g")
            .config("spark.executor.cores", "4")
            .config("spark.sql.shuffle.partitions", "400")
            .config("spark.sql.adaptive.enabled", "true")
            .config("spark.serializer",
                    "org.apache.spark.serializer.KryoSerializer")
        )

    return builder.getOrCreate()


# Использование
env = os.getenv("SPARK_ENV", "local")
spark = create_spark_session("etl-pipeline", env=env)

# Текущая конфигурация
for key, val in spark.sparkContext.getConf().getAll():
    print(f"{key} = {val}")

# Версия Spark
print(spark.version)

# Приложение и UI
print(spark.sparkContext.applicationId)
print(spark.sparkContext.uiWebUrl)

# Статус Executor-ов
print(spark.sparkContext.statusTracker().getExecutorInfos())

# Сколько памяти выделено
print(spark.sparkContext.getConf().get("spark.executor.memory"))