Spark — اسنیپت‌های منتخب

دفترچه‌ی میدانی یه مهندس داده: هر دستور PySpark جواب یه دام واقعی تو production رو میده — skew، small files، lineage که منفجر شده، NULLهایی که جوین نمی‌خورن، count() که ۱٫۲ ترابایت رو اسکن می‌کنه — و با یه خروجی واقعی کنسول هم تأییدش می‌کنه (df.show، پلن explain، متریک‌های MERGE). یه PySpark اندازه‌گیری‌شده و ابزارگذاری‌شده (سرعت ×۳۴ و ×۸۸)، با محوریت Delta Lake، پنجره‌بندی زمانی و کیفیت داده.

20 اسنیپت منتخب

• →پنجره زمانی واقعی با rangeBetween روی timestampروی ۲۴ ساعت گذشته به‌صورت لغزان تجمیع می‌کنی (نه N ردیف آخر) با تبدیل timestamp به ثانیه.
• →Forward fill: کشیدن آخرین مقدار معلوم به جلوبا last(ignorenulls=True) روی پنجره‌ای که از سمت گذشته بی‌کرانه، جاهای خالی یک سری رو پر می‌کنی — همون ffill پانداس ولی توزیع‌شده.
• →Sessionization: برش session بر اساس فاصله بی‌فعالیتیهر وقت فاصله دو رویداد از ۳۰ دقیقه بیشتر بشه، یک شناسه session جدید می‌دی — با جمع تجمعی روی یک فلگ شروع.
• →Semi-join: فیلتر بر اساس وجود، بدون آوردن ستون اضافهleft_semi جدول چپ رو بر اساس وجود کلید توی جدول راست فیلتر می‌کنه و حتی اگه سمت راست ردیف تکراری داشته باشه، هیچ ردیفی چندبرابر نمیشه.
• →اتصال null-safe با eqNullSafeتساوی معمولی NULLهای هر دو طرف رو کنار می‌ذاره (نتیجه مقایسه NULL با NULL خودش NULLه)؛ ولی eqNullSafe اون‌ها رو با هم تطبیق میده — برای کلیدهای ترکیبی nullable حیاتیه.
• →Salting: شکستن skew توی یک joinسمت جدول skewدار یک salt تصادفی اضافه می‌کنی و سمت مرجع explode می‌کنی، تا یک کلید بیش‌ازحد غالب روی N پارتیشن پخش بشه.
• →بازنویسی پویای پارتیشن‌ها (backfill امن)حالت dynamic فقط همون پارتیشن‌هایی رو که توی DataFrame در حال نوشتن هستن جایگزین می‌کنه، نه کل جدول رو — همون تنظیمی که backfillها رو نجات میده.
• →Bucketing: حذف shuffle از joinهای تکرارشوندهجدول‌هایی که روی همون کلید و همون تعداد bucket، bucket شدن: اتصال‌ها و تجمیع‌های بعدی دیگه نیازی به shuffle ندارن.
• →جایگزینی UDF پایتونی با توابع بومییک UDF پایتونی هر ردیف رو به یک worker پایتون سریال می‌کنه و Catalyst رو کور می‌کنه؛ ولی همون منطق با توابع بومی هنوز بهینه‌پذیر می‌مونه و ۱۰ تا ۱۰۰ برابر سریع‌تره.
• →aggregate: جمع‌کردن یه array توی یه مقدارfold بومی روی یه آرایه: جمع وزن‌دار، الحاق شرطی — بدون explode/groupBy و در نتیجه بدون shuffle.
• →SCD Type 2 با Delta MERGEتاریخچه‌سازی با یه حرکت دوگانه: نسخه فعلی رو می‌بندی و نسخه جدید رو درج می‌کنی، با ترفند mergeKey برابر NULL برای ردیف‌هایی که تغییر کردن.
• →checkpoint: بریدنِ یه lineage که از کنترل خارج شدهتوی pipelineهای تکرارشونده، plan منطقی اون‌قدر بزرگ میشه که کارِت به StackOverflow یا بازمحاسبه انفجاری می‌کشه؛ checkpoint این شجره رو می‌بُره.
• →خوندن execution plan: ۴ نشونه‌ای که باید دنبالش بگردیخروجی explain(mode='formatted') و نشونگرهایی که مهمن: Exchange (یعنی shuffle)، نوع join، PushedFilters و PartitionFilters.
• →monotonically_increasing_id: یکتا آره، پیوسته نهاین شناسه شماره پارتیشن رو توی بیت‌های بالاش کد می‌کنه: برای همین مقادیر بین پارتیشن‌ها میلیاردی می‌پرن. row_number سراسری پیوسته‌ست ولی همه‌چیزو روی یه task سریال می‌کنه.
• →از snapshotهای روزانه تا جریان CDC (شامل INSERT / UPDATE / DELETE)یه full_outer join بین snapshotهای امروز و دیروز، با مقایسه‌ی hash از نوع sha2 روی ستون‌های پایش‌شده، هر ردیفو INSERT یا UPDATE یا DELETE برچسب می‌زنه: یه جریان CDC واقعی، بدون کانکتور.
• →شمارش روزهای کاهش پیاپی (streak)همون الگوی gaps-and-islands، این‌بار روی روندها: یه جمع تجمعی از «گسست‌ها» سری رو گروه‌بندی می‌کنه و اندازه‌ی گروه‌های نزولی، محصولاتی که N روزه توی سقوطن رو لو می‌ده.
• →نگهداشت (retention) بر اساس کوهورت ماهانههر مشتری به کوهورت خودش (اولین ماه خرید) وصل میشه، سن به ماه pivot میشه و ستون‌های M1/M2/M3 نگهداشت رو به‌صورت درصدی از کوهورت اولیه نشون میدن.
• →پیش‌تجمیع قبل از join: تقسیم shuffle بر ۱۶ هزاراگه ۸۰۰ میلیون کلیک رو قبل از join به ۵۰ هزار ردیف تجمیع کنی، یک SortMergeJoin سنگین تبدیل میشه به BroadcastHashJoin: همون نتیجه با shuffle تقریبا صفر.
• →محافظ ضدانفجار: برآورد یک join قبل از پرداخت هزینه‌اشتعداد ردیف‌هایی که یک join تولید می‌کنه رو میشه از قبل حساب کرد: جمع حاصل‌ضرب کاردینالیتی‌ها به‌ازای هر کلید. یک آستانه، این برآورد رو تبدیل می‌کنه به یک محافظ که جلوش رو می‌گیره.
• →جایگزینی count() با isEmpty: محافظی که ۳ دقیقه هزینه می‌داداگه وجود داده رو با count() چک کنی، همه‌چیز رو اسکن می‌کنه تا یک عدد بسازه که فقط با صفر مقایسه‌اش می‌کنی؛ ولی isEmpty (یا take(1)) با پیداکردن اولین ردیف وایمیسه.