kakao·Aug 6, 2026·4 min readCurated summary

Experience Building and Operating a Personalized Airflow Testing Environment

Kakao’s data engineering team built AirZone to make Airflow DAG testing faster, easier, and safer across an ecosystem containing thousands of DAGs and multiple Hadoop clusters. Existing approaches required local setup, repeated Git synchronization, file copying, VPN access, or risky testing on production Airflow. AirZone instead creates an isolated, production-like Airflow environment for each pull request, managed through GitHub comments and Kubernetes automation. ## Limitations of Existing Testing Methods - **Local Airflow** - Requires configuring Airflow, Hadoop authentication, connections, and Docker locally. - Has a high initial setup cost and may differ from production. - **Development Airflow** - Requires committing and pushing every code change. - Git submodule updates and DAG parsing introduce long feedback delays. - **Test Airflow with SSH** - Allows files to be copied directly into a container. - Still requires copying files after every edit. - Access to production Hadoop requires connecting to a production VPN. - **Testing on production Airflow** - Heavy test DAGs consume shared scheduler, worker, and node resources. - A resource-intensive test can delay or interrupt unrelated projects. - Per-user isolation is therefore essential. ## AirZone Requirements - Provide an Airflow environment without requiring users to understand Kubernetes or Helm. - Allow code editing through a browser using Jupyter Notebook. - Execute DAGs against Hadoop and authentication mechanisms similar to production. - Create an independent environment for each pull request. - Prevent one user’s tests from affecting other workflows. ## PR-Based, Isolated Architecture - GitHub pull request comments serve as the user interface. - Users can create or delete an environment directly from a PR. - The resulting environment link is posted back to the PR. - Each PR receives a dedicated Kubernetes namespace based on the repository and PR number. - Airflow web server, scheduler, PostgreSQL, Jupyter, DAG volumes, and logs are isolated. - Multiple PRs can be tested simultaneously. - Cleanup is straightforward because the namespace defines the environment boundary. - A dedicated AirZone Helm chart packages the complete test environment. - Production-only components such as PGBouncer and external database connections are omitted where unnecessary. - Airflow, PostgreSQL, DAG storage, Jupyter, authentication, TLS, and logging are deployed together. ## Separating Requests from Deployment - `airzone-api` only validates requests: - Confirms that the PR exists and is open. - Checks branch information. - Prevents duplicate namespaces. - Kubernetes Jobs perform the long-running work: - Install the Helm release. - Run health checks. - Handle creation and deletion independently from the API process. - Job names include the operation and namespace, such as: - `create-airzone-{namespace}` - `delete-airzone-{namespace}` - Failed Jobs can be removed and recreated for retries. - Independent Job logs and status make deployment failures easier to diagnose. - A daily CronJob removes environments that remain after their PRs are closed. ## Building the Airflow Environment Each Helm deployment includes the components needed for a realistic test environment: - **Git integration:** Synchronizes the PR’s head repository and branch. - **DAG PVC:** Lets the scheduler and Jupyter use the same working directory. - **Airflow configuration:** Uses KubernetesExecutor and test-specific DAG scanning, logging, and Hadoop settings. - **Authentication:** Injects user and shared principals, keytabs, Jupyter tokens, and TLS certificates. - **Infrastructure placement:** Selects suitable node groups and a storage class in the same region. - **Centralized logging:** Connects Airflow logs to Elasticsearch and Kibana. - **Hadoop execution:** Existing infrastructure runs Hadoop tasks in dedicated pods using custom Hadoop images, Kerberos initialization, Spark, and Hive. ## Notifications and Security - KakaoWork sends: - An initial notification when a request is received. - A completion notification after deployment. - Operational error alerts. - Sensitive information, including Jupyter and Kubernetes namespace tokens, is not posted in public PR comments. - Tokens are delivered through KakaoWork instead, keeping authentication data separate from the broader PR audience. AirZone’s main recommendation is to make testing a disposable, reproducible environment tied to the pull request itself. By combining per-PR Kubernetes namespaces, Helm-based deployment, asynchronous Jobs, production-like Hadoop access, and automatic cleanup, teams can test DAGs quickly without burdening shared Airflow or production resources.

Read original(opens in new tab)

Apache SparkKubernetesPostgreSQLHelm+3

line·Jun 26, 2026·4 min readCurated summary

Solving the Cold-Start Problem in Search Reranking Through Embedding Stabilization: A LINE Part Time Jobs Case Study

LY Corporation improved LINE Part Time Jobs’ real-time search reranking by stabilizing user and item embeddings produced by a two-tower recommendation model. The approach addressed both cold-start degradation and daily embedding-space drift without changing the underlying model or training pipeline. Offline and online evaluations showed substantial gains, including a 4.7% overall KPI increase and 6.5% revenue growth. ## Search Reranking at LINE Part Time Jobs - Search consists of: - Retrieval, which finds listings matching a query. - Reranking, which orders the retrieved candidates. - The previous system ranked listings by cosine similarity between precomputed user-to-item two-tower embeddings. - This approach was computationally simple and captured broad user preferences, but: - It ignored query-specific information, such as the distance from a selected station. - Its embeddings combined behavior from multiple services and recommendation modules, not just search activity. - The team therefore introduced a dedicated real-time reranking model. ## Challenges with the Dedicated Reranking Model ### Cold Start - Most job listings are replaced at the beginning of each month. - New listings initially lack sufficient interaction data. - As a result, reranking quality dropped until enough training data accumulated. ### Embedding-Space Drift - Two-tower models were regularly retrained from random initialization. - Each training run produced a different embedding space. - Using embeddings as downstream features caused a mismatch between training-time and inference-time data, reducing model performance. ## Stabilizing the Embedding Space - Each day’s embeddings are aligned with the previous day’s stabilized embeddings. - The first day’s embeddings are used without stabilization. - This preserves continuity across retraining cycles and allows embeddings generated on different days to remain comparable. - Downstream models and embedding generation no longer need perfectly synchronized update schedules. ### Low-Rank SVD - User and item embeddings are converted into a more standardized low-dimensional representation. - Instead of decomposing the enormous user-item score matrix directly, transformation matrices are derived from the embedding matrices. - This makes the procedure practical for large-scale data. ### Orthogonal Procrustes Alignment - The transformed embeddings are aligned to the previous day’s stabilized space. - The orthogonal transformation only rotates or reflects the space. - Distances and inner-product relationships are therefore largely preserved, maintaining the two-tower model’s scoring behavior. ## Scalable Implementation - The algorithm was implemented with Apache Spark to handle LINE Part Time Jobs’ large datasets. - For low-rank SVD: - The original QR decomposition was optimized using Cholesky decomposition. - The Gram matrix \(G=A^\top A\) is decomposed to obtain the same upper-triangular matrix \(R\) as QR decomposition. - For Procrustes alignment: - The large matrix multiplication \(M=B^\top A\) is distributed across Spark. - The resulting \(e \times e\) matrix is small enough for SVD on a single node using NumPy. ## Evaluation Results ### Embedding Stability - Before stabilization, embeddings from randomly selected days had correlations close to zero. - After stabilization: - Similarity remained around 0.88 after one week. - Similarity remained around 0.87 after one month. - This reduced performance loss caused by embedding drift. ### Offline Evaluation - Unstabilized embeddings reduced nDCG by approximately 1–5% when training and inference used different days. - Stabilized embeddings improved: - Conversion nDCG by about 9.0%. - Click nDCG by about 4.5%. ### Online A/B Test - Search-page KPIs alone did not show statistically significant improvement. - Across the entire service: - KPIs increased by 4.7%. - Revenue increased by 6.5%. - The results suggest that the embeddings captured long-term user preferences that influenced later actions across the service, not only behavior on the search page. - The added embedding features also helped mitigate the initial cold-start problem. ## Practical Benefits and Future Work - The solution required no changes to the two-tower model itself. - Stabilization was added as post-processing, minimizing changes to existing pipelines and reducing deployment risk. - LY Corporation plans to test the method as the service expands its sources of job listings and to reuse the approach across other services through its internal machine-learning platform. Overall, sequential low-rank SVD and orthogonal Procrustes alignment provide a relatively simple way to make frequently retrained embeddings reliable downstream features while improving real-time reranking and business outcomes.

Read original(opens in new tab)

Apache SparkMachine LearningOnline LearningTwo Tower Embedding+4

line·Jun 23, 2026·4 min readCurated summary

Total Capacity Exceeds 1 EB! How Do You Connect Two HDFS Systems with Different Histories? Challenges and Design Decisions in Data Platform Integration

LY Corporation’s Tech-Verse 2026 article examines how its former LINE and Yahoo Japan organizations operated HDFS platforms exceeding one exabyte in total capacity. Although both platforms used Hadoop at scale, their access models, namespace architectures, permission systems, and operational practices differed substantially. The article argues that large-scale data platforms must be designed around actual usage patterns, not just storage capacity, and previews how the two environments were later connected after organizational integration. ## Different Operating Models - Former LINE built a unified analytics environment for broad, cross-departmental data use. - Users accessed data through a web portal that managed catalogs, permissions, and role-based approval workflows rather than interacting directly with HDFS or Apache Ranger. - BI tools, reporting systems, and ETL pipelines supported diverse use cases, but integrating multiple existing clusters made operations complex. - Former Yahoo Japan evolved from a limited-purpose Hadoop deployment into a company-wide platform. - Its user interfaces and access methods were intentionally restricted, making the system easier to stabilize and support. - Yahoo Japan retained HDFS-style POSIX permissions, which limited flexibility compared with newer data-governance models. ## Different HDFS Architectures - Both platforms split their storage across multiple namespaces to overcome NameNode scaling limitations. - Each namespace used two to four NameNodes for redundancy, but the way namespaces were exposed differed: - LINE used **ViewFS**, requiring clients to maintain mount-table configurations. - Yahoo Japan used **Router-Based Federation (RBF)**, allowing routers to direct requests to the correct NameNode. - LINE shared DataNodes across namespaces, improving resource efficiency but increasing operational complexity. - LINE also had mixed NameNode and DataNode versions because several legacy platforms had been consolidated. - Yahoo Japan’s RBF design included Observer NameNodes to distribute read load. - These architectural differences affected later federation work, including connection endpoints, client configuration, network reachability, and permission management. ## Capacity and Network Challenges at LINE - Data growth exceeded forecasts, causing HDFS capacity shortages before new servers could be delivered. - Older servers were temporarily reused, leading to frequent node additions and removals. - Large changes in node count triggered HDFS Balancer activity and block redistribution, generating substantial network traffic. - Network engineers therefore had to coordinate closely with the Hadoop operations team during infrastructure changes. ## NameNode Metadata and Small-File Problems - As file and block counts increased, NameNode heap usage and processing load grew. - Larger heaps also increased garbage-collection times, making NameNodes slower and less stable. - The team analyzed regularly dumped FSImage data stored in Hive tables to identify users, paths, file counts, block counts, and data volumes. - They prioritized tables containing many small files where file compaction could significantly reduce block counts without requiring data deletion or schema changes. - File merging reduced both NameNode metadata pressure and the number of HDFS operations, improving response times for jobs. ## Namespace-Specific Load Patterns - Different namespaces experienced different types of pressure. - Temporary-file namespaces saw frequent Spark staging-file creation and deletion, producing repeated metadata updates requiring NameNode write locks. - When HDFS Balancer moved blocks, read-lock activity increased and could delay file creation and deletion. - Increasing Balancer parallelism initially worsened contention. - The team reduced parallelism to a level compatible with available DataNode disk capacity, balancing migration speed against cluster impact. ## Connecting the Two Platforms - Organizational integration introduced additional challenges beyond storage: - Determining which platform and entry point users should access - Reconciling different permission-management models - Establishing data-transfer paths between platforms - LINE’s ViewFS model depends on correctly distributed client mount tables. - Yahoo Japan’s RBF model depends on reliable, scalable, and reachable router infrastructure. - These differences directly influence cross-platform data movement, including transfers using DistCP. Large HDFS environments should be managed according to real workload behavior, namespace characteristics, and operational dependencies. Capacity planning alone is insufficient; teams should monitor metadata growth, small-file patterns, lock contention, balancing traffic, network effects, and the distinct access models of each platform.

Read original(opens in new tab)

Apache SparkData PlatformHdfsApache Hadoop+3

netflix·Jun 19, 2026·3 min readCurated summary

Data Projects: Managing Data Assets at Netflix Scale

Data Projects address Netflix’s difficulty managing millions of data assets and tens of thousands of workloads as teams and employees change. They replace asset-level permissions and human-owned workload identities with project-level grants and durable, synthetic identities. This makes access easier to maintain, workflows more resilient, and newly created assets easier to organize automatically. ## The Limits of Asset-Level Permissions - Netflix historically managed access through individual ACLs on each table. - Organizational changes required updating hundreds or thousands of permissions manually. - This overwhelmed support teams and encouraged overly broad access, such as granting access to the entire company. - The model did not scale with frequent reorganizations, team changes, and ownership transfers. ## The Limits of Human-Owned Workloads - Scheduled jobs and asynchronous workloads traditionally ran under the identity of their author. - When that person changed roles or left Netflix, the workload’s permissions changed or disappeared. - Reassigning the job to another employee often introduced new permission gaps. - This created a recurring “permissions whack-a-mole” across tens of thousands of business-critical workflows. ## Data Projects as a Management Container - A Data Project groups related tables, workflows, secrets, and other assets under one logical umbrella. - Teams manage permissions for the project instead of maintaining ACLs across every individual asset. - Grants can be assigned to users, groups, applications, and CI jobs. - Roles such as Contributor and Viewer define read/write or read-only access at the project level. ## Durable Project Identities - Each project receives a Netflix application identity and, optionally, an AWS IAM role. - Scheduled workloads execute as the project rather than as an individual employee. - The IAM role supports AWS use cases such as Spark jobs on Amazon EMR. - Privileged project members can assume the project identity from laptops or notebooks for testing and troubleshooting. - This provides a development context that matches the identity used in production. ## Gravity and Automatic Asset Organization - Assets created by workloads running under a project identity are automatically added to that project. - For example, tables created by a Maestro workflow become project assets without extra configuration. - This “gravity” keeps related outputs organized and makes future access and discovery easier. - Newly created assets inherit the project’s access model rather than requiring separate permissions. ## Securing Maestro Workflows - Maestro runs ETL pipelines, data movement jobs, machine-learning training, and other batch workloads. - As a Trusted Workload Manager, Maestro can mint identity tokens for scheduled executions. - A single workflow may be checked against table ACLs, Netflix resource policies, and AWS IAM policies. - Using a durable project identity prevents failures caused by changes to the original author’s account. - Project-scoped secrets also remain available when ownership changes. Data Projects provide Netflix with a scalable foundation for access control, workload execution, and asset ownership. Moving management from individual assets and employees to durable, team-owned projects makes the platform more stable, auditable, and resilient to organizational change.

Read original(opens in new tab)

Apache SparkAccess ControlIdentity ManagementAws Iam+4

netflix·Jun 19, 2026·3 min readCurated summary

The Evolution of Cassandra Data Movement at Netflix

Netflix replaced its monolithic Cassandra-to-Iceberg connector, Casspactor, with a layered data movement engine built around direct reads from Cassandra backups in Amazon S3. Casspactor handled about 1,200 jobs and 3 PB daily but suffered from fragile metadata dependencies, skewed-partition failures, excessive intermediate tables, and limited support for higher-level data models. The new architecture uses Spark DataFrames and reusable, data-model-aware connectors to improve reliability, scalability, and cost efficiency. ## Casspactor’s Role and Limitations - Casspactor moved Cassandra data into Apache Iceberg using SSTables and metadata stored in S3 backups. - It supported critical Netflix workloads, including Member, Billing, Recommendations, and Subscriptions. - Its metadata view depended on several independent systems, each with different failure modes and update schedules. - Metadata could become inconsistent with actual backups, causing stale or incorrect data to be processed. - Cassandra maintenance or node replacement could break an entire region’s movement jobs because all nodes had to snapshot at the same clock second. ## Constraints for Higher-Level Data Abstractions - Cassandra-backed abstractions such as Key Value and Time Series inherited Casspactor’s limitations. - Large or skewed partitions caused executor memory failures and out-of-memory crashes. - Casspactor had no awareness of application-level data models, forcing downstream connectors to reconstruct them through costly post-processing. - Multiple intermediate Iceberg and snapshot tables increased storage costs and operational complexity. - Its backup composition model prevented reliable time travel to earlier backups after topology or keyspace schema changes. - The monolithic connector could not serve as a reusable foundation for specialized connectors. ## Direct S3 Metadata as the Source of Truth - The new design reads backup metadata directly from the S3 storage layer. - This removes the chain of external metadata dependencies. - Backup existence and completeness are determined from the files that actually contain the data. - Direct backup access also enables restoration of historical backup states. ## A Layered Connector Architecture - The Cassandra Analytics Wrapper builds on open-source Cassandra Analytics and Netflix’s internal backup format. - It uses an S3 client to read Cassandra backup files and convert them into standard Spark DataFrames. - A Connector Factory, implemented through Java UDFs and transforms, lets each abstraction define its own optimized connector. - Key Value, Time Series, and other models can transform generic DataFrames according to their own semantics. - Improvements to the shared reading engine automatically benefit every connector. ## Performance and Operational Improvements - Mutation compaction and processing run at Spark executor level, allowing better handling of wide and highly skewed partitions. - Reduced data shuffling helps prevent memory failures on large datasets. - Direct DataFrame output eliminates costly intermediary Iceberg tables. - Automatic job sizing adjusts resource usage based on source-table characteristics, reducing manual tuning. - Fewer dependencies improve reliability and make the system easier to maintain. Netflix’s new engine provides a shared, backup-native foundation while keeping data-model-specific logic in separate connectors. This approach is better suited to expanding Cassandra abstractions and large-scale data movement than maintaining another monolithic connector.

Read original(opens in new tab)

Apache SparkAmazon S3Apache IcebergApache Cassandra+3

toss·Jun 19, 2026·5 min readCurated summary

Spark Connect on Kubernetes #1: Building a Robust Spark Connect

Toss Securities operates Spark Connect as a production service on Kubernetes so analysts and engineers can use Spark without complex setup. Spark Connect replaces per-application Drivers with long-running servers, making clients lighter and sessions faster, but it also introduces shared-failure and resource-contention problems. The post argues that production reliability requires both reducing server-wide failure triggers and distributing sessions across multiple replicas. ## How Classic Spark Works - Spark consists of: - A **Driver**, which plans jobs, schedules tasks, and collects results. - **Executors**, which perform the distributed computations. - In Classic Spark: - **Client mode** runs the Driver inside the client process. - **Cluster mode** launches the Driver in the cluster for each submitted application. - Both modes assume that one application has one Driver and one workload. - Clients also need Spark libraries, JVM support, and configuration. ## What Spark Connect Changes - Spark Connect turns the Driver into a pre-started, long-running server. - Clients send unresolved logical plans encoded with Protocol Buffers over gRPC. - The server handles analysis, optimization, scheduling, and execution. - Results are streamed back using Arrow. - This resembles a database accessed through JDBC. ### Benefits - **Thin clients:** Clients do not need the full Spark runtime or JVM. - **Language and platform independence:** Notebooks, BI tools, SQL clients, and different programming languages can use the same server. - **Fast session creation:** Sessions connect to an already-running server. - **Better client-failure tolerance:** A disconnected notebook does not necessarily terminate server-side work. ## Problems Created by Shared Long-Running Servers Spark’s internal design often assumes “one application equals one workload.” Sharing one application across many users breaks that assumption. ### A Shared Driver Becomes a Single Point of Failure - Multiple sessions share one `SparkContext` and Driver JVM. - A Driver failure terminates all sessions, jobs, and caches attached to it. - Spark’s global `spark.executor.maxNumFailures` counter can shut down the entire application after enough executor failures. - Because all sessions contribute to the same counter, one user’s unstable or memory-intensive query can terminate unrelated users’ workloads. - The counter is global, persists over time, and is separate from per-query task-level fault tolerance such as `spark.task.maxFailures`. ### Resource Contention and Scheduling Limits - `newSession()` isolates SQL state and namespaces, but not CPU, memory, or executors. - Heavy workloads can occupy all task slots and delay smaller queries. - FIFO scheduling favors earlier jobs, and Spark does not preempt tasks already using slots. - Fair Scheduler pools can influence task-slot ordering, but cannot provide true CPU or memory isolation. - Spark Connect does not automatically propagate `spark.scheduler.pool` to the server-side execution thread, causing queries to fall into the default pool unless the server explicitly assigns pools. - Actual resource isolation must therefore be implemented outside Spark’s task scheduler. ### Fixed Server Capacity - A server’s image, Driver and Executor resources, and Spark configuration are fixed when it starts. - Dynamic Resource Allocation can adjust executor counts, but cannot change the server’s basic specification. - Flexible scaling and team-level isolation require creating or replacing servers, which is addressed in a later part of the series. ## Reducing Server-Wide Failures Before adding replicas, Toss Securities reduces the chance that one bad query can kill the shared server. - Set `spark.executor.maxNumFailures` effectively high enough to disable the global shutdown mechanism. - Use `spark.executor.failuresValidityInterval` to periodically clear accumulated failure records. - Rely on query-scoped controls: - `spark.task.maxFailures` stops tasks that repeatedly fail due to OOMs or exceptions. - `spark.stage.maxConsecutiveAttempts` stops jobs whose stages repeatedly fail, such as from shuffle-fetch errors. - These limits must be tuned carefully: overly aggressive values can cause healthy queries to fail during temporary infrastructure problems. - With this approach, executor failures terminate the problematic query rather than the entire Spark Connect server. ## Protecting Driver Memory from Large Results - Spark Connect streams query results through the Driver, so a large `collect()` can threaten Driver memory. - `spark.driver.maxResultSize` aborts an action when accumulated task results exceed the configured limit. - The limit is checked before large executor-side results are fetched into Driver memory. - The default 1 GB value assumes a single workload; in a multi-session server, it should be reduced or tuned based on the number of concurrent queries. ## Replicating Spark Connect Servers - Configuration alone cannot prevent Driver OOMs, node failures, or other catastrophic events. - The stronger isolation boundary is a separate SparkContext. - Multiple identical Spark Connect replicas are deployed: - Each replica has its own Driver, SparkContext, and Executors. - A failure affects only the sessions assigned to that replica. - Other replicas can continue accepting sessions. - Replica-based deployment reduces the blast radius from the entire Spark Connect service to an individual server instance. ## Practical Recommendation For a multi-user Spark Connect service, disable global executor-failure shutdown, enforce query-level failure limits, protect Driver memory with `spark.driver.maxResultSize`, and use multiple replicas to contain unavoidable Driver or node failures. Scheduler pools can improve ordering, but they should not be treated as true resource isolation.

Read original(opens in new tab)

Apache SparkKubernetesGrpcFault Tolerance+4

slack·May 5, 2026·3 min readCurated summary

From SSH to REST: A Security-Driven Modernization of Slack’s EMR Data Pipelines

Slack had more than 700 SSH-based operators running critical EMR workloads, creating security risks, operational failures, and barriers to infrastructure modernization. The company replaced these connections with REST-based job submission across eight data regions without downtime. YARN Distributed Shell was the key enabler for migrating arbitrary command-line jobs that lacked dedicated REST APIs. ## How Slack’s SSH Architecture Developed - Airflow originally connected directly to EMR master nodes using `SSHOperator`. - Over time, teams created more than 700 SSH-based jobs for: - Spark and MapReduce workloads - AWS CLI commands - Custom Python scripts - Data-transfer operations such as `hadoop distcp` - The approach was simple but tightly coupled orchestration workers to production clusters. ## Security and Operational Costs of SSH - Direct SSH access expanded the attack surface. - SSH keys had to be distributed and rotated across orchestration workers. - Auditing required correlating activity across multiple systems. - Permissions became complicated, often involving custom security groups and configurations. - Jobs ran on EMR master nodes, causing resource contention. - Restarted Kubernetes pods could break SSH connections. - Long-running processes could become orphaned “zombie” jobs. - Connection failures made job success or failure difficult to determine. - SSH dependencies blocked Spark-on-Kubernetes, EMR on EKS, AWS child-account migration, and better observability. - Slack’s search-indexing pipeline was especially sensitive because it processed terabytes of data daily and supported search for millions of users. ## REST-Based Job Submission - SSH creates a stateful connection whose failure can leave job status ambiguous. - REST APIs provide a durable, server-managed lifecycle: - `POST` submits a job and returns an ID. - `GET` retrieves its status. - `DELETE` cancels it cleanly. - Clients can crash or restart without terminating the underlying job. - Existing systems such as YARN, Trino, and Snowflake use this model. - YARN provides REST submission for Hadoop, Spark, Hive, and MapReduce workloads, but not arbitrary shell commands. ## YARN Distributed Shell - Spark and Hive already had REST-compatible options through Livy and HiveServer2. - The difficult cases were MapReduce and more than 300 CLI-based jobs. - Slack considered custom wrapper services, Ansible or Salt, and creating a new YARN job type. - These alternatives added complexity, security work, or long-term maintenance. - YARN Distributed Shell—implemented through `ApplicationMaster`—could execute arbitrary scripts inside YARN containers. - It used existing YARN APIs and authentication mechanisms, avoiding a custom security layer. ## The Distributed Shell Workflow - Upload a command script to S3, such as an `aws s3 sync` operation. - Submit a YARN application specifying: - The Distributed Shell application master - The S3 script location - Script metadata such as length and timestamp - YARN then: - Allocates a resource-managed container - Downloads and executes the script - Enforces memory and vCore limits - Provides isolation, retries, cancellation, and centralized logging By using REST submission and YARN Distributed Shell, Slack could remove SSH from its EMR data pipelines while preserving support for both standard data-processing jobs and arbitrary command-line workloads.

Read original(opens in new tab)

Apache SparkKubernetesRest ApiApache Airflow+3

line·Apr 3, 2026·4 min readCurated summary

From Hive to Iceberg: The Secret to 12x Faster Data Reflection

LINE Plus replaced a full-dump ETL pipeline for product data with incremental processing using Apache Iceberg and Apache Flink. The previous HBase/Hive workflow rewrote hundreds of millions of rows for every update, causing high compute costs and delays that left data up to an hour out of date. With the new architecture, update intervals were reduced from 60 minutes to 5 minutes—roughly a 12× improvement—while preserving consistency and fault tolerance. ## Limitations of Full-Data ETL - The existing HBase and Hive pipeline continuously collected CDC data in HDFS but had to merge it with existing data and rewrite the entire table before changes became queryable. - This caused: - High compute and storage costs - Dependence on limited shared Hadoop resources - Delayed updates and stale data - Snapshot-based extraction provides consistency, but large snapshots can take hours and retain old versions through MVCC, increasing system overhead. - Processing only the changed rows would reduce the workload from hundreds of millions of records to tens of thousands, separating update cost from total dataset size. ## Introducing Apache Iceberg - Iceberg manages data through metadata and table snapshots rather than relying solely on directory structures like traditional Hive tables. - It supports row-level `upsert` and `delete` operations. - This allows incremental changes to be written without rewriting the entire table, making much shorter ETL intervals possible. ## Requirements for the Streaming Pipeline The team evaluated Spark and Flink against three essential requirements: - **Data freshness:** Late-arriving compensation or replay data must not overwrite newer records. - **End-to-end exactly-once processing:** Iceberg updates and Kafka status messages must not partially succeed. - **Fault tolerance and state management:** Processing state must survive failures and restarts. A Kafka message indicating that all CDC data through a specific timestamp—such as 13:03—has been applied serves as the signal that a bulk extraction can safely begin. This requires complete confidence that the message accurately represents the Iceberg table’s committed state. ## Why Two-Phase Commit Was Necessary - Iceberg and Kafka are independent systems, so writing to one while failing to write to the other could create inconsistent state. - Two-phase commit (2PC) prevents partial success: - Both systems prepare their writes. - They commit only when all required operations succeed. - Any failure causes the operation to roll back. - Exactly-once processing also prevents duplicate or missing records during retries, network failures, or node restarts. - Together, these guarantees make Kafka status messages a reliable representation of the Iceberg table’s state. ## Choosing Flink over Spark - Spark Structured Streaming uses a micro-batch model, which makes fine-grained event-time and state control more difficult. - Flink provides native event-by-event streaming and better support for the required consistency model. - The team used Flink state to track each record’s `updatedate`: - Older late-arriving events are ignored. - Replayed historical data cannot overwrite newer values. - Flink checkpoints: - Persist streaming state externally. - Enable recovery from the latest consistent point. - Integrate with the Kafka sink’s 2PC mechanism. - Kafka messages remain in a pre-commit state until the Iceberg write and checkpoint both succeed. ## Kubernetes Deployment Options - The team compared: - **Native Kubernetes:** Requires manually configuring roles, service accounts, services, routing, deployments, slots, and jobs. - **Flink Kubernetes Operator:** Represents Flink infrastructure and jobs as custom resources, automating configuration such as routing and the web UI through Helm values. - Although Flink has greater operational complexity and a steeper learning curve than Spark, it was selected because it was the only option that satisfied all three core requirements at the engine level. The recommended architecture is an incremental Iceberg pipeline powered by Flink, with stateful processing, checkpoints, and two-phase commit between Iceberg and Kafka. This approach keeps data current, avoids expensive full-table rewrites, and provides reliable recovery and consistency at a five-minute update interval.

Read original(opens in new tab)

Apache SparkDatabase DesignKubernetesApache Iceberg+4

pinterest·Feb 17, 2026·4 min readCurated summary

Drastically Reducing Out-of-Memory Errors in Apache Spark at Pinterest

Pinterest developed **Auto Memory Retries** to reduce Spark out-of-memory failures without permanently assigning oversized executors to every task. The system detects OOM failures and retries affected tasks with progressively larger resource profiles, reducing both on-call incidents and wasted compute. Instead of tuning every job for its peak memory demand, Pinterest can size jobs around typical usage while handling exceptional tasks elastically. ## Pinterest’s Spark Environment - Pinterest processes more than **90,000 Spark jobs daily** across tens of thousands of nodes. - Its infrastructure includes: - Kubernetes clusters - Spark 3.2, with Spark 3.5 adoption underway - Apache Celeborn for shuffle - Apache YuniKorn for scheduling - Apache Gluten and Meta’s Velox for acceleration - Archer, Pinterest’s internal submission service - More than **4.6% of job failures** were caused by OOM errors. ## Why Manual Memory Tuning Was Insufficient - Pinterest’s clusters are memory-bound, so simply increasing executor sizes is expensive and difficult. - Automatic tuning generally reduces executor memory to match historical usage and improve resource efficiency. - Manual tuning can work, but requires substantial expertise because: - Different stages perform different operations. - Individual tasks may have very different memory needs because of data skew. - Configurations that work for most tasks may fail for a small number of high-memory tasks. - Auto Memory Retries allow jobs to target approximately their **P90 memory usage**, while automatically giving unusually demanding tasks more capacity. ## How Spark Executor Memory Works - An executor’s memory and CPU capacity determine how many tasks can run concurrently. - By default, each CPU core provides a task slot. - For example, with `spark.task.cpus=2`, an executor with two usable task slots and 8 GB of memory provides roughly 4 GB per task on average. - Memory is shared, so one task may temporarily use more than its average allocation if another uses less. - An OOM occurs when the combined memory usage of concurrent tasks exceeds the executor’s available memory. ## Auto Memory Retries Design Pinterest modified Spark’s scheduling loop so individual tasks can use resource profiles different from their parent `TaskSet`. - Each task can store an optional `taskRpId` identifying its retry resource profile. - Pinterest creates immutable retry profiles at **2x, 3x, and 4x** the base profile. - If off-heap memory is enabled, it is scaled as well. - Retries use a hybrid strategy: - **First retry:** Double `cpus per task`, allowing the task to run on an existing executor with fewer concurrent tasks. - **Later retry:** Launch a physically larger executor if the task still fails or already requires the entire executor. - The approach prioritizes reusing existing executors before provisioning larger ones. ## Changes to Spark Internals Pinterest extended core Spark components through Pinterest-specific subclasses rather than using a listener-only implementation. - **Task** - Stores the optional task resource profile ID. - **TaskSetManager** - Tracks tasks with non-default profiles. - Assigns the next larger retry profile after an OOM. - **TaskSchedulerImpl** - Allows tasks with increased CPU requirements to run on standard executors. - **ExecutorAllocationManager** - Tracks pending tasks by retry profile. - Requests larger executors when physical memory is required. - The feature-specific classes are loaded only when Auto Memory Retries is enabled. - The Spark UI was updated to display each task’s resource profile ID. ## Handling Tasks After an OOM - When a task fails on an executor with more than one core, its first retry doubles `spark.task.cpus`. - Other tasks in the same stage or future stages are unaffected. - Spark cannot reliably determine which concurrent task caused the executor-level OOM. - As a result, Pinterest treats **all tasks running on the terminated executor** as having failed due to OOM and routes them to retries that do not share the executor with other tasks. ## Practical Conclusion Pinterest’s approach makes executor sizing elastic at the task level: configure jobs for normal memory usage, then progressively increase resources only for tasks that need them. This can reduce OOM-related failures and operational load while avoiding the cost of running every task on oversized executors.

Read original(opens in new tab)

Apache SparkKubernetesMemory ManagementResource Allocation+4

pinterest·Feb 5, 2026·3 min readCurated summary

Next Generation DB Ingestion at Pinterest

Pinterest replaced fragmented, batch-oriented database ingestion with a unified Change Data Capture (CDC) framework. The new architecture uses Debezium/TiCDC, Kafka, Flink, Spark, and Iceberg to process only changed records, reducing latency from over 24 hours to minutes while lowering infrastructure costs. It also provides native row-level deletion, scalable operations, and improved compliance. ## Problems with the Legacy System - Batch workflows often delayed updates by more than 24 hours. - Full-table processing was inefficient because many tables changed by less than 5% each day. - Lack of row-level deletion support complicated data compliance. - Multiple independently maintained pipelines created operational complexity and inconsistent data quality. ## Unified CDC-Based Architecture - Supports MySQL, TiDB, and KVStore. - Captures database changes through a generic CDC service and publishes them to Kafka, typically in under one second. - Flink processes events in near real time and stores them in append-only CDC Iceberg tables on S3. - Spark jobs run periodically—often every 15 minutes—to merge recent changes into base Iceberg tables. - A bootstrap pipeline initializes base tables from historical database dumps. - Maintenance jobs handle compaction and snapshot expiration. - The framework is designed for at-least-once processing, petabyte-scale data, thousands of pipelines, and YAML-based configuration. ## CDC Tables and Base Tables - CDC tables act as time-series ledgers containing every change event. - CDC data typically becomes available within five minutes. - Base tables mirror the current state of the source database while retaining historical records. - Base-table latency is generally between 15 minutes and one hour. ## Upserting Changes into Base Tables - Spark first identifies the newest event for each primary key. - Events are ranked by timestamp and GTID, then deduplicated. - Iceberg’s `MERGE INTO` applies the resulting changes: - Deletes matching records when the event represents a deletion. - Updates existing records. - Inserts new records unless the event is a deletion. - The process uses a recent CDC window and a processing watermark to avoid reprocessing unnecessary data. ## Choosing Merge-on-Read - Pinterest standardized on Iceberg’s Merge-on-Read (MOR) strategy. - Copy-on-Write (COW) was rejected for most workloads because: - It requires more computation during writes. - It produces substantially larger replacement files, increasing storage costs. - MOR better balances update performance and storage efficiency for frequent incremental changes. ## Partitioning for Faster Upserts - Large base tables can be partitioned using a hash bucket of the primary key. - For example, `bucket(100, id)` distributes records across 100 partitions. - This allows Spark to process partitions in parallel and reduces the data scanned or rewritten during merges. - Iceberg tables are configured with format version 2, identifier fields, merge-on-read update and delete modes, and target file sizes. ## Small-File Challenge - Bucketing improved parallelism but caused each upsert to generate many small files within partitions. - The article indicates that Pinterest investigated this bottleneck and introduced further optimizations, though the supplied excerpt ends before describing them. Pinterest’s CDC-based design provides a substantially faster and more efficient alternative to full-table batch ingestion. Teams adopting a similar system should combine incremental CDC processing with partitioning, merge-on-read storage, bootstrapping, and ongoing file-maintenance strategies.

Read original(opens in new tab)

Apache SparkApache KafkaMysqlApache Iceberg+4

pinterest·Jan 13, 2026·3 min readCurated summary

PinLanding: Turn Billions of Products into Instant Shopping Collections with Multimodal AI

PinLanding is a production pipeline for turning billions of products into searchable shopping collections using multimodal AI. Rather than relying mainly on historical queries or manual curation, it derives structured product attributes from images and metadata, then aligns those attributes with real user search behavior. The system combines multimodal LLMs, embedding-based consolidation, a CLIP-style classifier, and distributed infrastructure to produce scalable, precise shopping feeds. ## Understanding Shopping Intent - Pinterest analyzes search history, autocomplete use, filters, and browsing paths to estimate shopping demand. - Existing systems handle high-volume queries such as “black cocktail dress” well, but provide weaker coverage for: - Long-tail queries - Conversational requests - Contextual intents such as “what to wear for an Italian summer vacation” - The analysis identifies: - Product areas with strong demand but poor collection coverage - Important attribute dimensions, including color, occasion, style, fit, price, and brand - The goal is to expand and improve collection coverage, not replace query understanding. ## Generating and Curating Shopping Topics - Each product is represented by an image plus metadata such as title, description, merchant tags, and price. - A vision-language model generates normalized key-value attributes rather than free-form descriptions. - Raw model output has high recall but produces: - Excessively specific attributes - Near-duplicates such as “boho,” “bohemian,” and “boho-chic” - Sparse attributes that apply to very few products - PinLanding builds a compact vocabulary through: - Frequency filtering to remove rarely useful attributes - Embedding-based clustering to merge semantically similar terms - Manual and LLM-assisted review - An LLM judge evaluates generated topics for semantic coherence, realistic shopping intent, and alignment with natural search phrasing. ## Scalable Attribute Assignment - Running the vision-language model over every product is too expensive and operationally fragile. - PinLanding trains a CLIP-inspired dual encoder: - One encoder embeds product images and text - Another embeds attribute phrases - Matching product-attribute pairs are trained as positives, while mismatches are negatives - A bidirectional contrastive loss aligns related products and attributes. - At inference, products and attributes are embedded once, and attributes are assigned when similarity exceeds a calibrated threshold. - This produces fewer distinct attributes while increasing the average number assigned to each product, creating a denser and more consistent attribute graph. ## Distributed Feed Construction - Ray handles large-scale batch inference across millions of products and topics. - The pipeline separates: - CPU-based image and metadata loading, tokenization, and serialization - GPU-based classifier inference - Streaming allows preprocessing and inference to overlap, while heterogeneous CPU and GPU clusters can scale independently. - The classifier pipeline reportedly completes in about 12 hours using eight NVIDIA A100 GPUs, at an estimated cost of roughly $500 per training run. - Feed construction uses approximate-nearest-neighbor techniques and strict attribute matching. - Topics are represented as attribute tuples, such as: - Category: dress - Color: yellow - Season: summer - Occasion: party - Apache Spark computes topic-product relevance using shared attributes and confidence weights, with partitioning and overlap filters reducing unnecessary candidate comparisons. The core recommendation is to combine user-behavior signals with content-first multimodal modeling. This approach can expand shopping coverage into conversational and long-tail intents while remaining practical through attribute consolidation, contrastive retrieval, and distributed inference.

Read original(opens in new tab)

Apache SparkMultimodal AIContrastive LearningClip+4

daangn·Dec 10, 2025Original article

Drawing a Karrot Data Map: (opens in new tab)

Daangn’s data governance team addressed the lack of transparency in their data pipelines by building a column-level lineage system using SQL parsing. By analyzing BigQuery query logs with specialized parsing tools, they successfully mapped intricate data dependencies that standard table-level tracking could not capture. This system now enables precise impact analysis and significantly improves data reliability and troubleshooting speed across the organization. **The Necessity of Column-Level Visibility** * Table-level lineage, while easily accessible via BigQuery’s `JOBS` view, fails to identify how specific fields—such as PII or calculated metrics—propagate through downstream systems. * Without granular lineage, the team faced "cascading failures" where a single pipeline error triggered a chain of broken tables that were difficult to trace manually. * Schema migrations, such as modifying a source MySQL column, were historically high-risk because the impact on derivative BigQuery tables and columns was unknown. **Evaluating Extraction Strategies** * BigQuery’s native `INFORMATION_SCHEMA` was found to be insufficient because it does not support column-level detail and often obscures original source tables when Views are involved. * Frameworks like OpenLineage were considered but rejected due to high operational costs; requiring every team to instrument their own Airflow jobs or notebooks was deemed impractical for a central governance team. * The team chose a centralized SQL parsing approach, leveraging the fact that nearly all data transformations within the company are executed as SQL queries within BigQuery. **Technical Implementation and Tech Stack** * **sqlglot:** This library serves as the core engine, parsing SQL strings into Abstract Syntax Trees (AST) to programmatically identify source and destination columns. * **Data Collection:** The system pulls raw query text from `INFORMATION_SCHEMA.JOBS` across all Google Cloud projects to ensure comprehensive coverage. * **Processing and Orchestration:** Spark is utilized to handle the parallel processing of massive query logs, while Airflow schedules regular updates to the lineage data. * **Storage:** The resulting mappings are stored in a centralized BigQuery table (`data_catalog.lineage`), making the dependency map easily accessible for impact analysis and data cataloging. By centralizing lineage extraction through SQL parsing rather than per-job instrumentation, organizations can achieve comprehensive visibility without placing an integration burden on individual developers. This approach is highly effective for BigQuery-centric environments where SQL is the primary language for data movement and transformation.

Apache SparkData GovernanceBigqueryApache Airflow+4

line·Dec 5, 2025Original article

Introducing a New A/B Testing System (opens in new tab)

LY Corporation has developed an advanced A/B testing system that moves beyond simple random assignment to support dynamic user segmentation. By integrating a dedicated targeting system with a high-performance experiment assigner, the platform allows for precise experiments tailored to specific user characteristics and behaviors. This architecture enables data-driven decisions that are more relevant to localized or specialized user groups rather than relying on broad averages. ## Limitations of Traditional A/B Testing * General A/B test systems typically rely on random assignment, such as applying a hash function to a user ID (`hash(id) % 2`), which is simple and cost-effective. * While random assignment reduces selection bias, it is insufficient for hypotheses that only apply to specific cohorts, such as "iOS users living in Osaka." * Advanced systems solve this by shifting from general testing across an entire user base to personalized testing for specific segments. ## Architecture of the Targeting System * The system processes massive datasets including user information, mobile device data, and application activity stored in HDFS. * Apache Spark is used to execute complex conditional operations—such as unions, intersections, and subtractions—to refine user segments. * Segment data is written to Object Storage and then cached in Redis using a `{user_id}-{segment_id}` key format to ensure low-latency lookups during live requests. ## A/B Test Management and Assignment * The system utilizes "Central Dogma" as a configuration repository where operators and administrators define experiment parameters. * A Test Group Assigner orchestrates the process: when a client makes a request, the assigner retrieves experiment info and checks the user's segment membership in Redis. * Once a user is assigned to a specific group (e.g., Test Group 1), the system serves the corresponding content and logs the event to a data store for dashboard visualization and analysis. ## Strategic Use Cases and Future Plans * **Content Recommendation:** Testing different Machine Learning models to see which performs better for a specific user demographic. * **Targeted Incentives:** Limiting shopping discount experiments to "light users," as coupons may not significantly change the behavior of "heavy users." * **Onboarding Optimization:** Restricting UI tests to new users only, ensuring that existing users' experiences remain uninterrupted. * **Platform Expansion:** Future goals include building a unified admin interface for the entire lifecycle of an experiment and expanding the system to cover all services within LY Corporation. For organizations looking to optimize user experience, transitioning from random assignment to dynamic segmentation is essential for high-precision product development. Ensuring that segment data is cached in a high-performance store like Redis is critical to maintaining low latency when serving experimental variations in real-time.

Apache SparkRedisData ArchitecturePersonalization+3

naver·Dec 1, 2025Original article

Naver TV (opens in new tab)

This technical session from NAVER ENGINEERING DAY 2025 explores the architectural journey of building a low-latency query system for real-time transaction reports. The project focuses on resolving the tension between high data freshness, massive scalability, and rapid response times for complex, multi-dimensional filtering. By leveraging Apache Iceberg in conjunction with StarRocks’ materialized views, the team established a performant data pipeline that meets the demands of modern business intelligence. ### Challenges in Real-Time Transaction Reporting * **Query Latency vs. Data Freshness:** Traditional architectures often struggle to provide immediate visibility into transaction data while maintaining sub-second query speeds across diverse filter conditions. * **High-Dimensional Filtering:** Users require the ability to query reports based on numerous variables, necessitating an engine that can handle complex aggregations without pre-defining every possible index. * **Scalability Requirements:** The system must handle increasing transaction volumes without degrading performance or requiring significant manual intervention in the underlying storage layer. ### Optimized Architecture with Iceberg and StarRocks * **Apache Iceberg Integration:** Iceberg serves as the open table format, providing a reliable foundation for managing large-scale data snapshots and ensuring consistency during concurrent reads and writes. * **StarRocks for Query Acceleration:** The team selected StarRocks as the primary OLAP engine to take advantage of its high-speed vectorized execution and native support for Iceberg tables. * **Spark-Based Processing:** Apache Spark is utilized for the initial data ingestion and transformation phases, preparing the transaction data for efficient storage and downstream consumption. ### Enhancing Performance via Materialized Views * **Pre-computed Aggregations:** By implementing Materialized Views, the system pre-calculates intensive transaction summaries, significantly reducing the computational load during active user queries. * **Automatic Query Rewrite:** The architecture utilizes StarRocks' ability to automatically route queries to the most efficient materialized view, ensuring that even ad-hoc reports benefit from pre-computed results. * **Balanced Refresh Strategies:** The research focused on optimizing the refresh intervals of these views to maintain high "freshness" while minimizing the overhead on the cluster resources. The adoption of a modern lakehouse architecture combining Apache Iceberg with a high-performance OLAP engine like StarRocks is a recommended strategy for organizations dealing with high-volume, real-time reporting. This approach effectively decouples storage and compute while providing the low-latency response times necessary for interactive data analysis.

Apache SparkApache IcebergData ArchitectureStarrocks+3

netflix·Sep 22, 2025Original article

Scaling Muse: How Netflix Powers Data-Driven Creative Insights at Trillion-Row Scale | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix’s Muse platform has evolved from a simple dashboard into a high-scale Online Analytical Processing (OLAP) system that processes trillions of rows to provide creative insights for promotional media. To meet growing demands for complex audience affinity analysis and advanced filtering, the engineering team modernized the data serving layer by moving beyond basic batch pipelines. By integrating HyperLogLog sketches for approximate counting and leveraging in-memory precomputed aggregates, the system now delivers low-latency performance and high data accuracy at an immense scale. ### Approximate Counting with HyperLogLog (HLL) Sketches To track metrics like unique impressions and qualified plays without the massive overhead of comparing billions of profile IDs, Muse utilizes the Apache Datasketches library. * The system trades a small margin of error (approximately 0.8% with a logK of 17) for significant gains in processing speed and memory efficiency. * Sketches are built during Druid ingestion using the HLLSketchBuild aggregator with rollup enabled to reduce data volume. * In the Spark ETL process, all-time aggregates are maintained by merging new daily HLL sketches into existing ones using the `hll_union` function. ### Utilizing Hollow for In-Memory Aggregates To reduce the query load on the Druid cluster, Netflix uses Hollow, an internal open-source tool designed for high-density, near-cache data sets. * Muse stores precomputed, all-time aggregates—such as lifetime impressions per asset—within Hollow’s in-memory data structures. * When a user requests "all-time" data, the application retrieves the results from the Hollow cache instead of forcing Druid to scan months or years of historical segments. * This approach significantly lowers latency for the most common queries and frees up Druid resources for more complex, dynamic filtering tasks. ### Optimizing the Druid Data Layer Efficient data retrieval from Druid is critical for supporting the application’s advanced grouping and filtering capabilities. * The team transitioned from hash-based partitioning to range-based partitioning on frequently filtered dimensions like `video_id` to improve data locality and pruning. * Background compaction tasks are utilized to merge small segments into larger ones, reducing metadata overhead and improving scan speeds across the cluster. * Specific tuning was applied to the Druid broker and historical nodes, including adjusting processing threads and buffer sizes to handle the high-concurrency demands of the Muse UI. ### Validation and Data Accuracy Because the move to HLL sketches introduces approximation, the team implemented rigorous validation processes to ensure the data remained actionable. * Internal debugging tools were developed to compare results from the new architecture against the "ground truth" provided by legacy batch systems. * Continuous monitoring ensures that HLL error rates remain within the expected 1–2% range and that data remains consistent across different time grains. For organizations building large-scale OLAP applications, the Muse architecture demonstrates that performance bottlenecks can often be solved by combining approximate data structures with specialized in-memory caches to offload heavy computations from the primary database.

Apache SparkMicroservicesGraphQLApache Druid+4