ANSI Mode: строгая типизация, implicit cast и совместимость

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit

spark = SparkSession.builder.getOrCreate()

sales_df = spark.createDataFrame([
    (1, 100, 0),
    (2, 200, 5),
    (3, 300, 10),
], ["order_id", "revenue", "clicks"])

result = sales_df.withColumn("cpc", col("revenue") / col("clicks"))
result.show()

+--------+-------+------+----+
|order_id|revenue|clicks| cpc|
+--------+-------+------+----+
|       1|    100|     0|null|
|       2|    200|     5|40.0|
|       3|    300|    10|30.0|
+--------+-------+------+----+

result.agg({"cpc": "avg"}).show()

+--------+
|avg(cpc)|
+--------+
|    35.0|
+--------+

campaign_df = spark.createDataFrame([
    (1, 2000000000, 500000000),
], ["campaign_id", "impressions", "extra"])

result = campaign_df.withColumn(
    "total",
    col("impressions").cast("int") + col("extra").cast("int")
)
result.show()

+-----------+-----------+------+-----------+
|campaign_id|impressions| extra|      total|
+-----------+-----------+------+-----------+
|          1| 2000000000|500000000|-1794967296|
+-----------+-----------+------+-----------+

spark.conf.set("spark.sql.ansi.enabled", "true")

spark.sql("SELECT 1 / 0").show()

java.lang.ArithmeticException: [DIVIDE_BY_ZERO]
Division by zero. Use try_divide to tolerate divisor being 0
and return NULL instead. If necessary set "spark.sql.ansi.enabled"
to "false" to bypass this error.
;  'Project [unresolvedstar()]
+- 'Project [(1 / 0) AS (1 / 0)#0]
   +- OneRowRelation

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ANSI Example") \
    .config("spark.sql.ansi.enabled", "true") \
    .getOrCreate()

spark.conf.set("spark.sql.ansi.enabled", "true")

spark.sql("SET spark.sql.ansi.enabled=true")

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("SELECT 10 / 0").show()

+---------+
|(10 / 0) |
+---------+
|     null|
+---------+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    spark.sql("SELECT 10 / 0").show()
except Exception as e:
    print(type(e).__name__, str(e)[:120])

AnalysisException [DIVIDE_BY_ZERO] Division by zero. Use try_divide to tolerate divisor being 0 and return NULL instead.

from pyspark.sql.functions import col

data = [(1, 100, 5), (2, 200, 0), (3, 300, 10)]
df = spark.createDataFrame(data, ["id", "revenue", "sessions"])

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    df.withColumn("rps", col("revenue") / col("sessions")).show()
except Exception as e:
    print(type(e).__name__, str(e)[:150])

SparkArithmeticException [DIVIDE_BY_ZERO] Division by zero. Use try_divide to tolerate divisor being 0 and return NULL instead.

from pyspark.sql.functions import try_divide

spark.conf.set("spark.sql.ansi.enabled", "true")

df.withColumn("rps", try_divide(col("revenue"), col("sessions"))).show()

+---+-------+--------+----+
| id|revenue|sessions| rps|
+---+-------+--------+----+
|  1|    100|       5|20.0|
|  2|    200|       0|null|
|  3|    300|      10|30.0|
+---+-------+--------+----+

from pyspark.sql.functions import try_divide, lit

df.withColumn("rps", try_divide(col("revenue"), col("sessions"), lit(0.0))).show()

+---+-------+--------+----+
| id|revenue|sessions| rps|
+---+-------+--------+----+
|  1|    100|       5|20.0|
|  2|    200|       0| 0.0|
|  3|    300|      10|30.0|
+---+-------+--------+----+

from pyspark.sql.functions import when, col

df.withColumn(
    "rps",
    when(col("sessions") > 0, col("revenue") / col("sessions"))
    .otherwise(None)
).show()

-- Legacy: не упадёт
SELECT order_id, amount / quantity AS unit_price FROM orders;

-- ANSI safe: явная защита
SELECT
    order_id,
    CASE WHEN quantity > 0
         THEN amount / quantity
         ELSE NULL
    END AS unit_price
FROM orders;

-- ANSI safe: через try_divide (Spark 3.3+)
SELECT order_id, try_divide(amount, quantity) AS unit_price FROM orders;

from pyspark.sql.functions import col, lit

spark.conf.set("spark.sql.ansi.enabled", "false")

overflow_df = spark.createDataFrame([(2147483647,)], ["max_int"])

result = overflow_df.withColumn(
    "overflow",
    col("max_int") + lit(1)
)
result.show()
result.printSchema()

+-----------+-----------+
|    max_int|   overflow|
+-----------+-----------+
| 2147483647|-2147483648|
+-----------+-----------+

root
 |-- max_int: integer (nullable = true)
 |-- overflow: integer (nullable = true)

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    result = overflow_df.withColumn("overflow", col("max_int") + lit(1))
    result.show()
except Exception as e:
    print(type(e).__name__, ":", str(e)[:100])

SparkArithmeticException : [INTEGER_OVERFLOW] integer overflow in 'Add', values: 2147483647, 1.

from pyspark.sql.functions import try_add, try_subtract, try_multiply

spark.conf.set("spark.sql.ansi.enabled", "true")

result = overflow_df.select(
    try_add(col("max_int"), lit(1)).alias("safe_add"),
    try_subtract(lit(-2147483648), lit(1)).alias("safe_sub"),
    try_multiply(col("max_int"), lit(2)).alias("safe_mul"),
)
result.show()

+--------+--------+--------+
|safe_add|safe_sub|safe_mul|
+--------+--------+--------+
|    null|    null|    null|
+--------+--------+--------+

from pyspark.sql.functions import col

df.withColumn(
    "total_clicks",
    col("daily_clicks").cast("long") * lit(365)
)

from pyspark.sql.types import DecimalType

df.withColumn(
    "revenue",
    col("amount").cast(DecimalType(18, 4))
)

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("SELECT '123' + 456").show()
spark.sql("SELECT true + 1").show()
spark.sql("SELECT 1.5 + 2").show()
spark.sql("SELECT '2024-01-01' + INTERVAL 1 DAY").show()

+-----------+
|(123 + 456)|
+-----------+
|        579|
+-----------+

+----------+
|(true + 1)|
+----------+
|         2|
+----------+

+---------+
|(1.5 + 2)|
+---------+
|      3.5|
+---------+

+---------------------------------------------+
|(CAST(2024-01-01 AS DATE) + INTERVAL '1' DAY)|
+---------------------------------------------+
|                                   2024-01-02|
+---------------------------------------------+

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("""
    SELECT
        '3.14' + 1 AS str_plus_int,
        'abc' + 1  AS bad_str_plus_int
""").show()

+------------+----------------+
|str_plus_int|bad_str_plus_int|
+------------+----------------+
|        4.14|            null|
+------------+----------------+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    spark.sql("SELECT '123' + 456").show()
except Exception as e:
    print(str(e)[:150])

[DATATYPE_MISMATCH.BINARY_OP_WRONG_TYPE] Cannot resolve '(CAST('123' AS DOUBLE) + 456)'
due to data type mismatch: argument 1 requires (numeric or interval day to second or
interval year to month or interval) type, however, '123' is of string type.

spark.sql("SELECT CAST('123' AS INT) + 456").show()

+------------------+
|(CAST(123 AS INT) + 456)|
+------------------+
|               579|
+------------------+

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("SELECT true + 1, false + 1, true * 5").show()

+----------+-----------+--------+
|(true + 1)|(false + 1)|(true*5)|
+----------+-----------+--------+
|         2|          1|       5|
+----------+-----------+--------+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    spark.sql("SELECT true + 1").show()
except Exception as e:
    print(str(e)[:100])

[DATATYPE_MISMATCH.BINARY_OP_WRONG_TYPE] Cannot resolve '(true + 1)' due to data type mismatch

spark.sql("SELECT CAST(true AS INT) + 1").show()
spark.sql("SELECT IF(flag, 1, 0) + base_count FROM events").show()

spark.conf.set("spark.sql.ansi.enabled", "true")

data = [("123",), ("abc",), ("456",), (None,), ("99.9",)]
df = spark.createDataFrame(data, ["raw_value"])

try:
    df.select("raw_value", col("raw_value").cast("int").alias("as_int")).show()
except Exception as e:
    print("CAST failed:", str(e)[:80])

df.select(
    "raw_value",
    col("raw_value").cast("int").alias("cast_int"),
).show()

from pyspark.sql.functions import try_cast

df.select(
    "raw_value",
    try_cast(col("raw_value"), "int").alias("safe_int")
).show()

+---------+--------+
|raw_value|safe_int|
+---------+--------+
|      123|     123|
|      abc|    null|
|      456|     456|
|     null|    null|
|     99.9|    null|
+---------+--------+

from pyspark.sql.functions import try_cast, col

dates_df = spark.createDataFrame([
    ("2024-01-15",),
    ("2024/02/30",),
    ("invalid-date",),
    ("2023-12-31",),
    ("",),
], ["raw_date"])

spark.conf.set("spark.sql.ansi.enabled", "true")

dates_df.select(
    "raw_date",
    try_cast(col("raw_date"), "date").alias("parsed_date")
).show()

+------------+-----------+
|    raw_date|parsed_date|
+------------+-----------+
|  2024-01-15| 2024-01-15|
|  2024/02/30|       null|
|invalid-date|       null|
|  2023-12-31| 2023-12-31|
|            |       null|
+------------+-----------+

-- SQL-синтаксис TRY_CAST
SELECT
    user_id,
    TRY_CAST(age_raw AS INT)             AS age,
    TRY_CAST(signup_date AS DATE)        AS signup,
    TRY_CAST(revenue_str AS DECIMAL(12,2)) AS revenue
FROM raw_events
WHERE TRY_CAST(age_raw AS INT) BETWEEN 18 AND 100;

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("""
    CREATE OR REPLACE TEMP VIEW products AS
    SELECT 'Very Long Product Name That Exceeds Limit' AS name
""")

spark.sql("""
    SELECT CAST(name AS VARCHAR(10)) AS short_name FROM products
""").show()

+----------+
|short_name|
+----------+
|Very Long |
+----------+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    spark.sql("""
        SELECT CAST('Very Long Product Name' AS VARCHAR(5)) AS s
    """).show()
except Exception as e:
    print(str(e)[:120])

[CAST_INVALID_INPUT] The value 'Very Long Product Name' of the type "STRING" cannot be cast to "VARCHAR(5)"
because it is malformed. Correct the value as per the syntax, or change its target type.

spark.conf.set("spark.sql.storeAssignmentPolicy", "ANSI")

from pyspark.sql.functions import col

arr_df = spark.createDataFrame([
    ([1, 2, 3],),
    ([10, 20],),
], ["numbers"])

spark.conf.set("spark.sql.ansi.enabled", "false")

arr_df.select(col("numbers")[5]).show()

+----------+
|numbers[5]|
+----------+
|      null|
|      null|
+----------+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    arr_df.select(col("numbers")[5]).show()
except Exception as e:
    print(str(e)[:100])

SparkArrayIndexOutOfBoundsException: [ARRAY_INDEX_OUT_OF_BOUNDS] The index 5 is out of bounds.
Array has 3 elements (indices from 0 to 2). Use the try_element_at function to avoid this error.

from pyspark.sql.functions import try_element_at

spark.conf.set("spark.sql.ansi.enabled", "true")

arr_df.select(
    "numbers",
    try_element_at(col("numbers"), lit(1)).alias("first"),
    try_element_at(col("numbers"), lit(5)).alias("fifth"),
).show()

+---------+-----+-----+
|  numbers|first|fifth|
+---------+-----+-----+
|[1, 2, 3]|    1| null|
| [10, 20]|   10| null|
+---------+-----+-----+

from pyspark.sql.functions import try_element_at, lit

map_df = spark.createDataFrame([
    ({"a": 1, "b": 2},),
], ["data"])

spark.conf.set("spark.sql.ansi.enabled", "true")

map_df.select(
    try_element_at(col("data"), lit("a")).alias("key_a"),
    try_element_at(col("data"), lit("z")).alias("key_z"),
).show()

+-----+-----+
|key_a|key_z|
+-----+-----+
|    1| null|
+-----+-----+

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("SELECT CAST('2024-13-01' AS DATE)").show()
spark.sql("SELECT CAST('2024-01-32' AS DATE)").show()

+----+
|null|
+----+

+----+
|null|
+----+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    spark.sql("SELECT CAST('2024-13-01' AS DATE)").show()
except Exception as e:
    print(str(e)[:120])

[CAST_INVALID_INPUT] The value '2024-13-01' of the type "STRING" cannot be cast to "DATE"
because it is malformed.

spark.conf.set("spark.sql.session.timeZone", "UTC")

spark.conf.set("spark.sql.datetime.java8API.enabled", "true")

from pyspark.sql.functions import to_date, to_timestamp, col

events_df = spark.createDataFrame([
    ("15-01-2024 10:30:00",),
    ("32-01-2024 00:00:00",),
    ("invalid",),
], ["raw_ts"])

events_df.select(
    "raw_ts",
    to_date(col("raw_ts"), "dd-MM-yyyy").alias("event_date"),
    to_timestamp(col("raw_ts"), "dd-MM-yyyy HH:mm:ss").alias("event_ts"),
).show()

+-------------------+----------+-------------------+
|             raw_ts|event_date|           event_ts|
+-------------------+----------+-------------------+
|15-01-2024 10:30:00|2024-01-15|2024-01-15 10:30:00|
|32-01-2024 00:00:00|      null|               null|
|            invalid|      null|               null|
+-------------------+----------+-------------------+

spark.conf.set("spark.sql.ansi.enabled", "false")

spark.sql("""
    SELECT 1 AS value, 2 AS table, 3 AS end
""").show()

+-----+-----+---+
|value|table|end|
+-----+-----+---+
|    1|    2|  3|
+-----+-----+---+

spark.conf.set("spark.sql.ansi.enabled", "true")

try:
    spark.sql("SELECT 1 AS end").show()
except Exception as e:
    print(str(e)[:150])

[PARSE_SYNTAX_ERROR] Syntax error at or near 'end'(line 1, col 12): SELECT 1 AS end

spark.sql("SELECT 1 AS `end`").show()

+---+
|end|
+---+
|  1|
+---+

-- Безопасный паттерн
SELECT
    `from`    AS source_date,
    `table`   AS table_name,
    `end`     AS end_date,
    `value`   AS metric_value
FROM legacy_table;

-- PostgreSQL: ошибка при делении на ноль
SELECT revenue / sessions FROM events WHERE campaign_id = 1;
-- ERROR: division by zero

-- Spark Legacy: null
-- Spark ANSI: то же поведение, что PostgreSQL - ошибка

-- Переносимый код для обеих систем:
SELECT CASE WHEN sessions > 0
             THEN revenue / sessions
             ELSE NULL
       END AS cpc
FROM events WHERE campaign_id = 1;

-- Trino: CAST('abc' AS INTEGER) → NULL (!) или ошибка зависит от настройки
-- Spark ANSI: исключение

-- Переносимый код:
SELECT TRY_CAST(raw_value AS INTEGER) AS parsed_int FROM raw_table;
-- TRY_CAST поддерживается в Trino, Spark 3.2+, Snowflake, BigQuery (TRY_CAST аналог - SAFE_CAST)

-- BigQuery
SELECT SAFE_DIVIDE(revenue, sessions) AS cpc FROM events;
SELECT SAFE_CAST(raw_date AS DATE) AS event_date FROM raw_events;

-- Spark ANSI эквивалент
SELECT TRY_DIVIDE(revenue, sessions) AS cpc FROM events;
SELECT TRY_CAST(raw_date AS DATE) AS event_date FROM raw_events;

from delta.tables import DeltaTable

spark.conf.set("spark.sql.ansi.enabled", "true")

spark.sql("""
    CREATE TABLE IF NOT EXISTS events_delta (
        event_id  BIGINT,
        user_id   INT,
        amount    DECIMAL(12, 4),
        event_ts  TIMESTAMP
    ) USING DELTA LOCATION '/data/events'
""")

spark.sql("""
    INSERT INTO events_delta VALUES (1, 999999999999, 100.50, '2024-01-15 10:00:00')
""")

[INSERT_COLUMN_ARITY_MISMATCH.NOT_ENOUGH_DATA_COLUMNS] Cannot write to 'events_delta',
as the number of user-specified column names does not match the data schema.

[STORE_ASSIGNMENT_VIOLATION] Assignment implicit type coercion is not allowed.
'user_id' requires int but 'expr' has type bigint.

spark.conf.set("spark.sql.ansi.enabled", "true")

spark.sql("""
    ALTER TABLE events_delta
    ADD COLUMNS (new_metric DOUBLE COMMENT 'New feature metric')
""")

spark.sql("""
    ALTER TABLE events_delta
    CHANGE COLUMN user_id user_id BIGINT
""")

spark.conf.set("spark.sql.catalog.iceberg", "org.apache.iceberg.spark.SparkCatalog")
spark.conf.set("spark.sql.catalog.iceberg.type", "hadoop")
spark.conf.set("spark.sql.ansi.enabled", "true")

spark.sql("""
    CREATE TABLE iceberg.db.events (
        id     BIGINT,
        amount DECIMAL(18, 4)
    ) USING iceberg
""")

spark.sql("""
    INSERT INTO iceberg.db.events VALUES (1, CAST('invalid' AS DECIMAL(18, 4)))
""")

spark.conf.set("spark.sql.ansi.enabled", "true")

import logging

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger("ansi_migration")

queries_to_test = [
    ("division_report", "SELECT revenue / sessions FROM events"),
    ("type_cast_etl", "SELECT '123' + visits FROM web_logs"),
    ("integer_math", "SELECT clicks * multiplier FROM campaigns"),
]

errors = []
for query_name, query in queries_to_test:
    try:
        spark.sql(query).limit(1).collect()
        print(f"✅ {query_name}: OK")
    except Exception as e:
        errors.append((query_name, str(e)[:200]))
        print(f"❌ {query_name}: {str(e)[:100]}")

print(f"\nTotal errors: {len(errors)}")
for name, err in errors:
    print(f"  {name}: {err}")

df.withColumn("cpc", col("cost") / col("clicks"))
df.withColumn("cpc", try_divide(col("cost"), col("clicks")))

df.filter(col("date_col") > "2024-01-01")
df.filter(col("date_col") > lit("2024-01-01").cast("date"))

df.withColumn("total", col("count") * lit(365))
df.withColumn("total", col("count").cast("long") * lit(365L))

df.withColumn("flag_int", col("is_active") + lit(0))
df.withColumn("flag_int", col("is_active").cast("int"))

df.withColumn("amount", col("raw_amount").cast("double"))
df.withColumn("amount", try_cast(col("raw_amount"), "double"))

def run_with_ansi(spark, query):
    """Запустить запрос в ANSI mode, восстановить исходный режим после."""
    original = spark.conf.get("spark.sql.ansi.enabled", "false")
    try:
        spark.conf.set("spark.sql.ansi.enabled", "true")
        return spark.sql(query)
    finally:
        spark.conf.set("spark.sql.ansi.enabled", original)

run_with_ansi(spark, "SELECT try_divide(revenue, sessions) FROM events").show()

import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def ansi_spark():
    return SparkSession.builder \
        .config("spark.sql.ansi.enabled", "true") \
        .getOrCreate()

def test_division_safety(ansi_spark):
    """Проверить, что деление не падает при нулевых делителях."""
    df = ansi_spark.createDataFrame(
        [(1, 100, 0), (2, 200, 5)],
        ["id", "revenue", "clicks"]
    )
    result = df.withColumn(
        "cpc",
        try_divide(col("revenue"), col("clicks"))
    )
    rows = result.collect()
    assert rows[0]["cpc"] is None
    assert rows[1]["cpc"] == 40.0

def test_no_overflow(ansi_spark):
    """Проверить, что LONG используется там, где INT может переполниться."""
    df = ansi_spark.createDataFrame([(2_000_000_000,)], ["large_int"])
    result = df.withColumn(
        "doubled",
        col("large_int").cast("long") * 2
    )
    assert result.collect()[0]["doubled"] == 4_000_000_000

spark.conf.set("spark.sql.ansi.enabled", "true")

spark.conf.set("spark.sql.storeAssignmentPolicy", "ANSI")

spark.conf.set("spark.sql.storeAssignmentPolicy", "STRICT")

spark.conf.set("spark.sql.storeAssignmentPolicy", "LEGACY")

spark.conf.set("spark.sql.datetime.java8API.enabled", "true")

spark.conf.set("spark.sql.session.timeZone", "UTC")

spark.conf.set("spark.sql.legacy.timeParserPolicy", "EXCEPTION")

{"event_id": "1", "user_id": "100", "amount": "29.99", "ts": "2024-01-15 10:30:00", "clicks": "5"}
{"event_id": "2", "user_id": "abc", "amount": "-1", "ts": "2024-13-01", "clicks": "0"}
{"event_id": "3", "user_id": "200", "amount": "invalid", "ts": "2024-01-20", "clicks": "10"}

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, try_cast, try_divide, current_timestamp,
    when, lit, count
)
from pyspark.sql.types import LongType, DoubleType, DateType, IntegerType

spark = SparkSession.builder \
    .appName("ANSI ETL Pipeline") \
    .config("spark.sql.ansi.enabled", "true") \
    .config("spark.sql.storeAssignmentPolicy", "ANSI") \
    .getOrCreate()

bronze_df = spark.createDataFrame([
    ("1", "100", "29.99", "2024-01-15 10:30:00", "5"),
    ("2", "abc", "-1",    "2024-13-01",          "0"),
    ("3", "200", "invalid", "2024-01-20",         "10"),
    ("4", "300", "150.00", "2024-01-21",          "3"),
], ["event_id", "user_id", "amount", "ts", "clicks"])

silver_df = bronze_df.select(
    try_cast(col("event_id"), LongType()).alias("event_id"),
    try_cast(col("user_id"), IntegerType()).alias("user_id"),
    try_cast(col("amount"), DoubleType()).alias("amount"),
    try_cast(col("ts"), DateType()).alias("event_date"),
    try_cast(col("clicks"), IntegerType()).alias("clicks"),
)

silver_with_cpc = silver_df.withColumn(
    "avg_revenue_per_click",
    try_divide(col("amount"), col("clicks").cast("double"))
)

good_rows = silver_with_cpc.filter(
    col("event_id").isNotNull() &
    col("user_id").isNotNull() &
    col("amount").isNotNull() &
    col("event_date").isNotNull()
)

bad_rows = silver_with_cpc.filter(
    col("event_id").isNull() |
    col("user_id").isNull() |
    col("amount").isNull() |
    col("event_date").isNull()
)

print("=== Good rows ===")
good_rows.show()

print("=== Bad rows (quarantine) ===")
bad_rows.show()

total = silver_with_cpc.count()
good = good_rows.count()
bad = bad_rows.count()
print(f"Quality report: {good}/{total} valid, {bad}/{total} quarantined ({bad/total*100:.1f}%)")

=== Good rows ===
+--------+-------+------+----------+------+---------------------+
|event_id|user_id|amount|event_date|clicks|avg_revenue_per_click|
+--------+-------+------+----------+------+---------------------+
|       1|    100| 29.99|2024-01-15|     5|                5.998|
|       4|    300|150.00|2024-01-21|     3|               50.000|
+--------+-------+------+----------+------+---------------------+

=== Bad rows (quarantine) ===
+--------+-------+------+----------+------+---------------------+
|event_id|user_id|amount|event_date|clicks|avg_revenue_per_click|
+--------+-------+------+----------+------+---------------------+
|       2|   null|  -1.0|      null|     0|                 null|
|       3|    200|  null|2024-01-20|    10|                 null|
+--------+-------+------+----------+------+---------------------+

Quality report: 2/4 valid, 2/4 quarantined (50.0%)

good_rows.write \
    .format("delta") \
    .mode("append") \
    .save("/silver/events")

bad_rows.withColumn("quarantine_ts", current_timestamp()) \
    .write \
    .format("delta") \
    .mode("append") \
    .save("/silver/events_quarantine")

spark.conf.set("spark.sql.ansi.enabled", "true")

spark.sql("INSERT INTO production_table SELECT * FROM etl_result")

raw_df.withColumn("amount", col("raw").cast("double"))

raw_df.withColumn("amount", try_cast(col("raw"), "double"))

df.withColumn("amount", try_cast(col("raw"), "double")) \
  .agg({"amount": "avg"})

df_with_types = df.withColumn("amount", try_cast(col("raw"), "double"))

null_rate = df_with_types.filter(col("amount").isNull()).count() / df.count()
if null_rate > 0.05:
    raise ValueError(f"Too many null amounts: {null_rate:.1%}")

df_with_types.agg({"amount": "avg"}).show()

# spark-defaults.conf
spark.sql.ansi.enabled=true  # ОПАСНО для shared cluster

df.withColumn("active_count", col("is_active") + col("total_count"))

df.withColumn("active_count", col("is_active").cast("int") + col("total_count"))

spark.conf.set("spark.sql.ansi.enabled", "true")

df.createOrReplaceTempView("events")
spark.sql("SELECT end, value, table FROM events")

spark.sql("SELECT `end`, `value`, `table` FROM events")

df.withColumnRenamed("end", "end_date") \
  .withColumnRenamed("value", "metric_value") \
  .withColumnRenamed("table", "source_table") \
  .createOrReplaceTempView("events")

spark.conf.set("spark.sql.ansi.enabled", "true")
spark.conf.set("spark.sql.storeAssignmentPolicy", "ANSI")
spark.conf.set("spark.sql.datetime.java8API.enabled", "true")
spark.conf.set("spark.sql.session.timeZone", "UTC")