# List files in HDFS
hdfs dfs -ls /user/data/

# Upload a file to HDFS
hdfs dfs -put localfile.csv /user/data/

# Read HDFS file in PySpark
df = spark.read.csv("hdfs://namenode:9000/user/data/sales.csv")

# Check replication factor for a file
hdfs fsck /user/data/myfile.parquet -files -blocks

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 10 \
  --executor-memory 4g \
  --executor-cores 2 \
  my_spark_job.py

-- Create external Hive table over HDFS Parquet data
CREATE EXTERNAL TABLE sales (
  order_id  STRING,
  amount    DECIMAL(10,2),
  city      STRING
)
STORED AS PARQUET
LOCATION '/data/sales/';

-- In PySpark, read the same Hive table directly
df = spark.sql("SELECT city, SUM(amount) FROM sales GROUP BY city")

# Read real-time data from Kafka into Spark Structured Streaming
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-host:9092") \
    .option("subscribe", "user-events") \
    .load()

# Kafka messages come as bytes — decode them
df = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, ArrayType

json_data = [
    ('{"user":"alice","tags":["python","spark"],"address":{"city":"Bangalore"}}',),
    ('{"user":"bob","tags":["java"],"phone":"9999999999"}',)  # different fields!
]
df = spark.createDataFrame(json_data, ["json_col"])

schema = StructType([
    StructField("user", StringType()),
    StructField("tags", ArrayType(StringType())),
    StructField("address", StructType([StructField("city", StringType())]))
])

parsed = df.withColumn("parsed", from_json(col("json_col"), schema))
parsed.select("parsed.user", "parsed.tags", "parsed.address.city").show()
# +-----+----------------+---------+
# |alice|[python, spark] |Bangalore|
# |  bob|        [java]  |     null|  ← missing fields become null

from pyspark.sql.functions import split, size, length

reviews = [
    ("1", "This product is absolutely amazing! Best purchase ever."),
    ("2", "Terrible quality. Broke after 2 days. Very disappointed."),
]
df = spark.createDataFrame(reviews, ["id", "review"])

# Basic processing on raw text
df = df.withColumn("word_count", size(split(col("review"), " "))) \
       .withColumn("char_count", length(col("review")))

# For NLP: use Spark NLP or Pandas UDF with transformers
from pyspark.ml.feature import Tokenizer, HashingTF
tokenizer = Tokenizer(inputCol="review", outputCol="words")
words_df = tokenizer.transform(df)

# Disk layout: each row stored together
# [Alice | 30 | 95000] [Bob | 25 | 72000] [Carol | 35 | 120000]

# Query: SELECT SUM(salary) FROM employees
# MUST read: name + age + salary for EVERY row
# Wasted: name + age columns read but never used!

# Disk layout: each column stored together
# [name col: Alice | Bob | Carol]
# [age col:  30    | 25  | 35]
# [salary:   95000 | 72000 | 120000]

# Query: SELECT SUM(salary) → ONLY reads salary block!
# Savings: name + age columns NEVER loaded from disk
# Also: similar data compresses much better (ints next to ints)

# Snappy: best balance of speed + ratio (default for Parquet)
df.write.parquet("s3://bucket/data/", compression="snappy")

# ZSTD: new standard, best ratio + speed combo
df.write.parquet("s3://bucket/data/", compression="zstd")

# Gzip: best ratio but slow and NOT splittable for CSV
df.write.csv("s3://bucket/data/", compression="gzip")

# Set default globally
spark.conf.set("spark.sql.parquet.compression.codec", "zstd")

# Parquet stores min/max statistics per row group (128MB chunks)
# Spark checks: does this row group overlap the filter range?
# If not → entire row group SKIPPED without reading

df = spark.read.parquet("s3://data/orders/")

result = df.filter(df.order_date >= "2024-01-01")

# Verify pushdown is happening:
result.explain(True)
# Look for: PushedFilters: [GreaterThanOrEqual(order_date,2024-01-01)]

# Column pruning (also pushed down)
# Only reads 'city' and 'amount' from Parquet, skips all other columns
df.select("city", "amount").filter(df.country == "IN").show()

# Original table: name, age only
df_v1 = spark.createDataFrame([("Alice", 30)], ["name", "age"])
df_v1.write.format("delta").save("s3://bucket/users")

# Business adds 'email' column later
df_v2 = spark.createDataFrame([("Bob", 25, "bob@example.com")], ["name", "age", "email"])

# mergeSchema: adds new columns automatically, no error!
df_v2.write.format("delta") \
     .option("mergeSchema", "true") \
     .mode("append") \
     .save("s3://bucket/users")

# Old rows get null for new column — clean merge
spark.read.format("delta").load("s3://bucket/users").show()
# |Alice|30|           null|  ← old row, null for new column
# |  Bob|25|bob@example.com|

# Default Java serialization — slow, large output
spark.conf.set("spark.serializer", "org.apache.spark.serializer.JavaSerializer")

# Kryo serialization — 3x faster, smaller (recommended for RDDs)
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

# Register custom classes for maximum Kryo performance
spark.conf.set("spark.kryo.classesToRegister", "com.example.MyClass")

# For DataFrames: Tungsten UnsafeRow is used AUTOMATICALLY
# Binary format with no Java object overhead
# This is why DataFrames are faster than RDDs!

# Delta Lake file structure
s3://bucket/my_table/
  ├── _delta_log/
  │   ├── 00000000000000000000.json   # version 0: CREATE TABLE
  │   ├── 00000000000000000001.json   # version 1: INSERT
  │   ├── 00000000000000000002.json   # version 2: UPDATE
  │   └── 00000000000000000010.checkpoint.parquet  # checkpoint every 10 commits
  ├── part-00000-abc.snappy.parquet   # actual data files (Parquet)
  └── part-00001-def.snappy.parquet

# Iceberg file structure
s3://bucket/my_iceberg_table/
  ├── metadata/
  │   ├── v1.metadata.json     # Table metadata + snapshot refs
  │   ├── snap-123.avro        # Manifest list (list of manifests)
  │   └── manifest-456.avro   # Manifest (list of data files + stats)
  └── data/
      └── country=IN/
          └── part-00000.parquet

# Time travel with Delta
spark.read.format("delta") \
     .option("versionAsOf", 5) \
     .load("s3://bucket/my_table")

# Time travel with Iceberg
spark.read.option("snapshot-id", 123456789) \
     .table("my_iceberg_table")