Big Data Analysis Skill

The numbers surface trade-offs. The decision belongs to the user.

⚙️ Engine: use Impala for exploration (seconds vs minutes). Switch to Hive only if the query requires features Impala lacks (e.g., LATERAL VIEW EXPLODE).

Mode 2: Coding — Rules That Prevent Silent Bugs

These rules exist because the resulting bugs are invisible: row counts look correct, no errors thrown, but data is silently wrong or performance collapses.

⚠️ Rule 0: DESCRIBE Before Writing Any Code

Never guess field names or types. A single type mismatch silently ruins JOINs.

hive -e "DESCRIBE db.table_name"
hive -e "SHOW CREATE TABLE db.table_name"

If JOIN key types differ across tables, cast explicitly — do not rely on implicit conversion.

⚠️ Rule 1: Never Hard-Code Table Names in Spark Source

Hard-coded names break multi-environment deployment and are impossible to override without recompiling.

// ❌
val df = spark.sql("SELECT * FROM some_db.some_table WHERE dt = '${dt}'")

// ✅ receive via CLI args, declare in .job config
val inputTable = cmd.getOptionValue("inputTable")
val df = spark.sql(s"SELECT * FROM $inputTable WHERE dt = '${dt}'")

⚠️⚠️ Rule 2: Keep Long-Text Fields Out of GROUP BY

User-generated text (descriptions, names) often contains ASCII control characters (\x01, \x03, etc. — Hive column delimiters). If such a field is in GROUP BY, Hive/Impala splits one row into many, shifting all downstream field values. Row count still looks normal — the bug is invisible.

Rule: finish all aggregation in CTEs, then append text fields in the outermost SELECT via LEFT JOIN.

-- In Hive/Impala:
WITH core AS (
    SELECT key_id, group_concat(cast(id AS string), ',') AS id_list
    FROM source_table
    GROUP BY key_id   -- ⚠️ no free-text fields here
)
SELECT core.*, regexp_replace(t.description, '[\\x00-\\x1f]', '') AS description
FROM core LEFT JOIN text_table t ON core.key_id = t.key_id AND t.dt = '${dt}';

Engine note: group_concat is Hive/Impala syntax. In Spark SQL, use concat_ws(',', collect_list(cast(id AS string))) instead. Mixing engines in one job is a common source of "function not found" errors.

Regex in Spark Scala: inside a Spark Scala s"""...""" string, use [\u0000-\u001f] instead of [\\x00-\\x1f] — Scala processes the backslash before SQL sees it. See references/spark-pitfalls.md §8.1.

⚠️⚠️ Rule 3: Filter First, Then Aggregate on Large Tables

Directly running GROUP BY + group_concat on a billion-row raw table causes OOM. First reduce the set in a CTE, then join the large table.

WITH targets AS (SELECT DISTINCT entity_id FROM ... WHERE conditions),
core AS (
    SELECT t.entity_id, group_concat(cast(d.id AS string), ',') AS id_list
    FROM targets t
    INNER JOIN huge_table d ON t.entity_id = d.entity_id
    GROUP BY t.entity_id   -- aggregates only the filtered subset
)
SELECT * FROM core;  -- this is a read query, not INSERT; see Rule 6 for INSERT patterns

⚠️⚠️⚠️ Rule 4: Use Spark SQL, Not the DataFrame API

Real project benchmark: DataFrame API (manual cache + repeated .count()) → 367 lines, 45+ Jobs, 3 hours. Spark SQL equivalent → 50 lines, 1 Job, 15 minutes.

Root causes:

• Every .count() triggers a full Spark Job
• foreachPartition(_ => ()) is a no-op — it does not materialize cache

// ✅
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")  // see Rule 5
spark.sql(s"INSERT OVERWRITE TABLE $output PARTITION(dt='${dt}') WITH ... SELECT ...")

If cache() is genuinely needed (result consumed by two independent branches), trigger it with a real Action:

df.count()                                          // ✅ always works; adds one extra Job
df.write.mode("overwrite").format("noop").save("")  // ✅ Spark 3.0+ with noop DataSource on classpath; no disk I/O
df.foreachPartition((_: Iterator[Row]) => ())       // ❌ no-op — never use this

Prefer count() unless you have confirmed noop DataSource is available in your cluster. If uncertain, ask the user.

Remove all debug count()/collect() calls before deploying to production — each one is a full extra Job.

⚠️⚠️ Rule 5: Control Broadcast JOIN Threshold Based on Table Size

Spark may auto-broadcast a table it considers "small" — if that estimate is wrong, you get task explosion and job timeout. Always set this before any SQL execution when working with tables of unknown or medium-to-large size:

// Set before any SQL runs — has no effect if set after query planning
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

If you observe a Stage stuck at BroadcastExchange in Spark UI with exploding task counts, disable immediately and rerun. See references/spark-pitfalls.md §3 for details and how to force-broadcast genuinely small tables with SQL hints.

⚠️⚠️⚠️ Rule 6: Never Use SELECT * in INSERT

CREATE TABLE IF NOT EXISTS does not modify an existing table's schema. INSERT ... SELECT * matches by position. Adding a new column shifts all subsequent columns silently — row count stays the same, data is corrupt.

// ❌
spark.sql(s"INSERT OVERWRITE TABLE $output PARTITION(dt='$dt') SELECT * FROM tmp")

// ✅ always list columns explicitly
spark.sql(s"""INSERT OVERWRITE TABLE $output PARTITION(dt='$dt')
             |SELECT col1, col2, col3, new_col FROM tmp""".stripMargin)

When adding a column to an existing table:

• Early stage: DROP TABLE IF EXISTS, then let the job recreate it with the new schema.
• Production: ALTER TABLE ADD COLUMNS, then update every INSERT statement. After the run, verify the new column has actual values — row count alone cannot detect a column shift.

Dynamic column matching (when schema may vary across runs):

val tableColumns = spark.sql(s"DESCRIBE $output")
    .filter("col_name NOT IN ('dt', '') AND col_name NOT LIKE '#%'")
    .select("col_name").as[String].collect().toSeq

val dataColumns = result.columns.toSeq
val missingCols = tableColumns.diff(dataColumns)
// ⚠️ order by dataColumns first — NEVER reorder by tableColumns
val selectExprs = dataColumns.map(c => s"`$c`") ++ missingCols.map(c => s"NULL AS `$c`")

result.createOrReplaceTempView("tmp_result")
spark.sql(s"""INSERT OVERWRITE TABLE $output PARTITION(dt='$dt')
             |SELECT ${selectExprs.mkString(", ")} FROM tmp_result""".stripMargin)

⚠️⚠️ Rule 7: Use LEFT JOIN for Optional Fields

An INNER JOIN to fetch an optional field (description, tag, title — any field not guaranteed to exist for every row) silently drops non-matching rows. Row count on the primary table looks fine; you only notice the loss if you compare before/after counts explicitly.

-- ❌ rows with no description are silently dropped
... INNER JOIN meta_table meta ON core.id = meta.id

-- ✅ missing metadata becomes NULL; no rows are lost
... LEFT JOIN meta_table meta ON core.id = meta.id AND meta.dt = '${dt}'

⚠️ Rule 8: Refresh Metadata After Writing

Spark writes to HDFS but does not update the Hive Metastore. Querying immediately after a write without refreshing returns stale or empty results.

spark.sql(s"REFRESH TABLE $output")   // or MSCK REPAIR TABLE / INVALIDATE METADATA in Hive/Impala

⚠️ Rule 9: UDF Type Safety — Return Types Only

The restriction is on return types, not parameter types. Seq[String] as a UDF input parameter is completely safe. Only nested Scala collection return types cause NoClassDefFoundError at runtime due to JVM type erasure.

// ✅ Safe input parameter — Seq[String] as argument is fine
val myUdf = udf((items: Seq[String], size: Int) => {
    // ❌ Unsafe return: Array[Seq[String]]
    items.grouped(size).map(_.toSeq).toArray        // crashes at runtime

    // ✅ Safe return: flatten each group with a delimiter
    items.grouped(size).map(_.mkString("|||")).toArray
})

See references/spark-pitfalls.md §5 for the full type-safety table and downstream split reconstruction pattern.

Quick Error Reference

Intermediate Table Hygiene

• Keep intermediate tables; configure a TTL/expiration policy so they are cleaned up automatically
  • Standard Hive: use a scheduled DROP TABLE or external lifecycle management tool
  • Some platforms support TBLPROPERTIES extensions for automatic expiration — consult your cluster's documentation
• Partition by dt; write one partition per run to support reruns
• Delete debug count()/collect() calls before production deployment

Reference

• references/spark-pitfalls.md — detailed root-cause analysis and extended code examples for Rules 1–9
• references/sql-patterns.md — SQL patterns where AI assistants commonly write incorrect code (Rules 2–3 and control character handling)