HintsToday

How Spark interacts with HDFS and performs in-memory computations while allowing data to be read from and written back to HDFS:

1. Reading Data from HDFS into Spark

• Data Retrieval: When Spark starts a job that requires reading data stored in HDFS, it reads the data from disk (HDFS) into Spark’s distributed memory (RAM) across the cluster nodes.
• Data Locality: Spark optimizes data retrieval by attempting to schedule tasks on the nodes where the data blocks reside. This minimizes data transfer across the network and speeds up processing.

2. In-Memory Computation in Spark

• RDDs (Resilient Distributed Datasets): When data is read from HDFS into Spark, it is typically loaded into RDDs or DataFrames. RDDs are distributed across the memory (RAM) of the cluster nodes.
• Transformations: Spark allows users to perform a variety of transformations (e.g., map, filter, join) on RDDs/DataFrames. These transformations are lazy, meaning they don’t compute results immediately but build a DAG (Directed Acyclic Graph) of operations to be executed.
• In-Memory Storage: Once data is loaded into RDDs or DataFrames, it resides in memory, allowing Spark to perform operations much faster than if it had to repeatedly read from disk.
• Caching and Persistence: Spark provides mechanisms to cache or persist RDDs/DataFrames in memory, so they can be reused across multiple actions (e.g., count, collect, saveAsTextFile) without being recomputed from scratch. You can choose different storage levels, such as memory-only or memory-and-disk, depending on the available resources.

3. Writing Data Back to HDFS

• Actions Trigger Execution: When an action like saveAsTextFile, saveAsTable, or write is called, Spark triggers the execution of the DAG, performing the transformations that were lazily defined.
• Data Shuffling (if necessary): During execution, some operations may require shuffling data across the network (e.g., groupBy, reduceByKey). This intermediate data is handled in memory but can spill to disk if necessary.
• Writing to HDFS: After all transformations are executed in memory, Spark writes the final output back to HDFS. Spark can write the results to HDFS in various formats such as text, Parquet, ORC, etc.

4. Example Workflow

Here’s a step-by-step example of how this process might look in practice:

Step 1: Read Data from HDFS

from pyspark.sql import SparkSession

# Initialize Spark session
spark = SparkSession.builder
    .appName("HDFS and In-Memory Example")
    .getOrCreate()

# Read data from HDFS
df = spark.read.csv("hdfs://namenode:9000/path/to/input.csv", header=True, inferSchema=True)

Step 2: In-Memory Computation

# Perform transformations
df_filtered = df.filter(df["age"] > 30)
df_grouped = df_filtered.groupBy("country").count()

# Optionally cache the DataFrame in memory
df_grouped.cache()

Step 3: Write Results Back to HDFS

# Write the result back to HDFS
df_grouped.write.csv("hdfs://namenode:9000/path/to/output.csv", header=True)

5. Why In-Memory Computation is Fast

• Reduced Disk I/O: Once the data is loaded into memory, Spark avoids repeated disk I/O operations, which are much slower than RAM access.
• Efficiency in Iterative Algorithms: In-memory storage is particularly beneficial for iterative algorithms (e.g., in machine learning), where the same dataset is processed multiple times.
• Reusability: Cached datasets can be reused across multiple operations without having to reload from HDFS, speeding up the overall computation.

6. Fault Tolerance and Spill to Disk

• Fault Tolerance: Even though data is processed in memory, Spark provides fault tolerance. If a node fails, the RDDs can be recomputed from their lineage (the sequence of transformations that created them) using the original data in HDFS.
• Spill to Disk: If the memory is insufficient to hold all the data (e.g., when handling very large datasets), Spark can spill data to disk temporarily. This ensures that jobs can still be completed, though with some performance penalty.

7. Writing Data Back to HDFS

Writing data back to HDFS ensures that the results of your computation are stored persistently and can be accessed later. This is a common practice in big data workflows where HDFS serves as the central storage system for processed data.

In summary, Spark efficiently reads data from HDFS into its distributed memory for fast, in-memory processing. While the initial data resides on disk (in HDFS), Spark performs computations in memory, significantly speeding up the processing. Once the computations are done, Spark can write the results back to HDFS. This combination of in-memory processing with persistent storage in HDFS provides a powerful and flexible framework for big data processing.

What happens in Hadoop if a node fails? how hadoop manages with lost block of Data?

Hadoop is designed with fault tolerance in mind, ensuring that data remains accessible even if a node in the cluster fails. Here’s how Hadoop handles node failures and the associated loss of data blocks:

1. HDFS Architecture Overview

• Data Blocks: In Hadoop’s HDFS (Hadoop Distributed File System), files are split into fixed-size blocks (default is 128 MB) and distributed across multiple nodes in the cluster.
• Replication Factor: Each data block is replicated across several nodes to ensure fault tolerance. The default replication factor is three, meaning each block is stored on three different nodes.

2. Node Failure and Block Loss Management

When a node fails, the blocks of data stored on that node may become temporarily inaccessible. Here’s how Hadoop manages this situation:

a. Detection of Node Failure

• Heartbeat Mechanism: DataNodes (the nodes that store data in HDFS) send periodic heartbeats to the NameNode (the master node that manages the metadata and directory structure of the file system).
• Timeout: If the NameNode does not receive a heartbeat from a DataNode within a specific timeframe, it marks that DataNode as dead or failed.

b. Re-Replication of Blocks

• Block Replication Monitoring: The NameNode constantly monitors the replication status of all data blocks in the cluster.
• Under-Replicated Blocks: When a DataNode fails, the NameNode identifies blocks that are now under-replicated (i.e., have fewer than the required number of replicas due to the node failure).
• Re-Replication Process: The NameNode triggers the replication of these under-replicated blocks to other healthy DataNodes in the cluster. This ensures that the replication factor is maintained, and data remains fault-tolerant.

c. Recovery of Lost Blocks

• No Data Loss: If the replication factor is properly maintained, there is no data loss when a node fails because the same blocks are already stored on other nodes.
• Data Block Reconstruction: If the cluster has sufficient storage capacity, the missing blocks are copied to other DataNodes, ensuring the data is fully replicated as per the desired replication factor.

3. Rebalancing the Cluster

• Load Balancing: After a node failure and the subsequent re-replication, the cluster might become unbalanced (i.e., some nodes might have more data blocks than others).
• Rebalancing Process: Hadoop provides a balancer utility that can redistribute blocks across the cluster to ensure that no single DataNode is overloaded.

4. Handling NameNode Failures

• Single Point of Failure: In older versions of Hadoop (pre-Hadoop 2.x), the NameNode was a single point of failure. If the NameNode failed, the entire HDFS would be inaccessible.
• High Availability (HA): In modern Hadoop versions, NameNode High Availability is implemented. Two or more NameNodes are set up in an active-passive configuration, with shared storage (e.g., using a Quorum Journal Manager). If the active NameNode fails, the passive NameNode takes over, ensuring continued access to the HDFS.

5. Example Scenario

Imagine a Hadoop cluster where a file is stored in HDFS with a replication factor of three. This file is split into several blocks, and each block is stored on three different DataNodes. If one of these DataNodes fails:

• The NameNode detects the failure through the absence of a heartbeat.
• The NameNode identifies all the blocks that were stored on the failed DataNode and notes that they are now under-replicated.
• The NameNode selects healthy DataNodes to replicate the missing blocks.
• The data is copied to these DataNodes, ensuring that the replication factor is restored.
• The cluster continues to operate without data loss, and users remain unaware of the node failure.

Summary

Hadoop ensures fault tolerance by replicating data blocks across multiple nodes. When a node fails, the NameNode quickly detects the failure, identifies under-replicated blocks, and re-replicates them to other nodes. This process ensures that data remains available and consistent, even in the event of hardware failures, maintaining the integrity of the distributed file system.

Here are some additional resources to enhance your PySpark learning journey:

• Apache Spark Documentation: https://spark.apache.org/documentation.html
• PySpark Tutorial: https://sparkbyexamples.com/pyspark-tutorial/
• Databricks Academy: https://www.databricks.com/learn/training/login (offers free courses and learning resources)

Together, the DAG Scheduler and Task Scheduler coordinate the parallel processing and execution of jobs in a Spark cluster. The DAG Scheduler focuses on dividing the job into stages, while the Task Scheduler manages the distribution and execution of individual tasks on the worker nodes.

How the stages are decided? What is the Stage Boundry?

In Spark, stages are logical units of execution that are formed based on transformations applied to a DataFrame or RDD. A stage is composed of a series of narrow transformations (operations that don’t require data movement across nodes) that can be executed without requiring data to be shuffled. A stage boundary is created whenever a wide transformation occurs, which involves shuffling data across the network.

What is Shuffling?

• Shuffling is the process of redistributing data across partitions, typically because a wide transformation (like groupBy, join, or reduceByKey) requires data to be grouped or combined in a specific way.
• Shuffling involves moving data between partitions across different worker nodes, which is an expensive operation in terms of both time and resources (network and disk I/O).

Stages and Shuffling:

1. Narrow Transformations: These transformations (e.g., map, filter, flatMap) do not require moving data between partitions. These operations can be pipelined within the same stage.
   • Example: If you’re applying map and filter transformations, they will be grouped together into the same stage because they operate independently on each partition.
2. Wide Transformations: These transformations (e.g., groupBy, reduceByKey, join) require shuffling data between partitions. This means the output of one transformation needs to be moved across nodes for the next operation to proceed. When a wide transformation happens, it causes a stage boundary.
   • Example: A groupBy transformation will require a shuffle because Spark needs to collect all the data for a particular key into the same partition to compute the result.

When Does a Stage Boundary Occur?

A stage boundary is created before shuffling happens. This means:

• The DAG Scheduler detects that a shuffle is required when encountering a wide transformation and splits the job into two stages:
  • Stage 1: Executes all narrow transformations up to the point where the wide transformation occurs.
  • Stage 2: The wide transformation creates the boundary and requires data to be shuffled. Once the shuffle is complete, Stage 2 executes.

How Stages Are Decided:

1. Narrow Transformations in the Same Stage:
   • Narrow transformations (e.g., map, filter, flatMap) are grouped together in a single stage. These transformations can be executed without data movement, as they only require access to the data within the same partition.
2. Wide Transformations Create Stage Boundaries:
   • A wide transformation (e.g., groupByKey, join, reduceByKey) requires data from multiple partitions to be reshuffled (moved across nodes). As soon as a wide transformation is encountered, the DAG Scheduler creates a new stage boundary.
   • Data is shuffled across the network between these stages.
3. Stages are Decided Based on Dependencies:
   • Narrow Dependency: A narrow dependency is when each partition of the parent RDD is used by at most one partition of the child RDD. These are pipelined within the same stage.
   • Wide Dependency: A wide dependency is when multiple partitions in the child RDD depend on the same partition in the parent RDD. This triggers a shuffle, and Spark creates a new stage.

Example of Stage Creation:

Let’s consider an example:

df = spark.read.csv("data.csv")
df_filtered = df.filter(df['value'] > 10)
df_mapped = df_filtered.withColumn("new_column", df['value'] * 2)
df_grouped = df_mapped.groupBy("category").sum("new_column")
df_grouped.show()

• Stage 1:
  • This stage will contain the filter and map operations (filter and withColumn), which are narrow transformations. These operations will be executed within the same stage because they operate independently on each partition, with no data movement required across partitions.
• Stage 2:
  • When the groupBy transformation is encountered, Spark realizes that it needs to shuffle the data across partitions to group the data by the category column. This creates a stage boundary.
  • Stage 1 will process the data and produce output that needs to be shuffled.
  • Stage 2 starts after the shuffle, which redistributes the data across nodes so that all rows with the same category end up in the same partition.

Flow of Execution:

1. Stage 1 (Before Shuffle):
   • Narrow transformations (filter, map) are executed.
   • The output is partitioned and prepared for the shuffle.
2. Shuffle:
   • The data is shuffled, meaning rows are redistributed across partitions based on the category column. This process includes sending data over the network to ensure all the rows with the same key (category) are in the same partition.
3. Stage 2 (After Shuffle):
   • The groupBy operation is executed on the shuffled data.
   • Tasks for this stage run on partitions where the shuffled data has been placed.

Important Points about Shuffling and Stages:

• Shuffling Happens Between Stages: The shuffle happens after Stage 1 completes, and it prepares data for Stage 2. The output from the shuffle becomes the input for the next stage.
• Stage Boundary Happens Before Shuffling: Spark creates a stage boundary when it encounters a wide transformation (e.g., groupBy), which necessitates a shuffle.
• Stages are Independent: Each stage is independent. After one stage completes and data is shuffled (if required), the next stage begins.
• Wide Transformation (Shuffle): A shuffle operation will result in data movement across the network, leading to the creation of a new stage. Examples of wide transformations that require a shuffle include:
  • groupByKey()
  • reduceByKey()
  • join()
  • distinct()
  • repartition()
  • coalesce()

When Does a Shuffle Happen?

A shuffle happens when Spark needs to redistribute data across partitions due to a wide transformation. This shuffle involves:

1. Writing intermediate data to disk.
2. Sending this data across the network to other nodes.
3. Reading the shuffled data into the correct partitions.

Summary of Stage Boundaries and Shuffling:

• Stages are divided based on the need for shuffling.
• Narrow transformations are grouped into the same stage, while wide transformations (that require a shuffle) create a new stage boundary.
• A shuffle happens between stages, not within a stage. Once a shuffle is triggered by a wide transformation, a new stage begins after the shuffle has completed.

In-memory computing (up to 100x faster than Hadoop)- so what it does differently than Hadoop, can you explain with example, and how memory in computing happens in spark and how not happens in Hadoop?

Great question — and this is a core interview topic. Let’s break it down clearly and compare Hadoop vs Spark from the execution and memory perspective with examples and internals.

🚀 Why Spark is “In-Memory” and 100x Faster Than Hadoop?

🔁 The Fundamental Difference

🧠 How Memory Computing Happens in Spark?

🔹 Spark: The Lazy, Memory-First Engine

• Spark creates RDDs or DataFrames, which are immutable, distributed collections.
• When you apply transformations (e.g., .filter(), .map()), Spark does not execute immediately.
• It builds a logical DAG (Directed Acyclic Graph) of all steps.
• Only when an action (e.g., .show(), .count(), .write()) is called:
  • Spark pipelines all the transformations.
  • Loads data into memory (RAM) on executors.
  • Processes it in-memory unless data size exceeds available RAM.
  • Only writes to disk (spill) if memory is insufficient.

✅ Intermediate steps are kept in memory, unless you force persistence on disk.

🔸 Hadoop MapReduce: Disk-Heavy by Design

• Each phase (Map → Shuffle → Reduce) is treated as a separate job.
• After the Map phase:
  • Intermediate key-value pairs are written to local disk.
  • Data is shuffled and sorted.
• Before the Reduce phase:
  • Data is read back from disk.
  • Then processed by reducers.
• No memory reuse, and no DAG of operations. It is just a pipeline of discrete steps.

📦 Visual Comparison

Hadoop MapReduce:

1. Map Task         → Output to disk
2. Shuffle & Sort   → Disk I/O
3. Reduce Task      → Final output written to HDFS

Total Disk I/O: 2x minimum

-------------------------------------

Spark DAG:

1. Transformation 1 (in-memory)
2. Transformation 2 (in-memory)
3. Action → Execute all in a pipeline
4. Write result to HDFS (only once)

Total Disk I/O: 1x (final result only)

💡 Example: Word Count

🐘 Hadoop Version:

// Mapper writes intermediate output to disk
// Reducer reads that from disk and writes final output to HDFS

This causes:

• Disk write after map
• Disk read before reduce
• Final disk write

⚡ Spark Version (PySpark):

rdd = spark.sparkContext.textFile("hdfs://input.txt")
word_counts = rdd.flatMap(lambda line: line.split()) \
                 .map(lambda word: (word, 1)) \
                 .reduceByKey(lambda a, b: a + b)
word_counts.saveAsTextFile("hdfs://output")

Here:

• All .flatMap, .map, and .reduceByKey happen in memory.
• Only the final .saveAsTextFile() writes to disk.
• Shuffle data may spill to disk only if it exceeds memory.

🧪 Optional: Forcing Spark to Spill (Just Like Hadoop)

You can configure Spark to spill to disk like Hadoop using these settings:

spark.memory.storageFraction = 0.3
spark.memory.fraction = 0.6
spark.shuffle.spill = true

✅ But by default, Spark tries RAM first, and spills only when needed — unlike Hadoop which always writes between phases.

🎯 Final Interview-Ready Takeaways

How memory tuning (executor memory, cache/persist) works in Spark? How to monitor memory/disk use from Spark UI?A Spark vs Hadoop performance benchmark walkthrough?

Let’s dive into all three of your points — these are critical for production readiness and frequently asked in Spark interviews. Each is explained in depth with examples, interview one-liners, and hands-on tips.

✅ 1. How Memory Tuning Works in Spark (Executor Memory, Cache/Persist)

🔧 Memory Structure in Spark Executors

Each executor has its memory divided like this:

[ Executor Memory ]
├── Reserved Memory (~300 MB)
├── User Memory (spark.memory.fraction - storageFraction)
├── Storage Memory (for cache, persist)
└── Execution Memory (for shuffles, joins, aggregations)

🔹 Key Configurations

➡️ So, for spark.executor.memory=4g:

• 60% (2.4g) is available for Spark
• 50% of that (1.2g) is for caching, rest for execution

🔁 Cache vs Persist

Example:

df = spark.read.csv("hdfs://largefile.csv")
df.cache()
df.count()  # Triggers cache to be used from here on

🚨 Tips

• Cache only when data will be reused.
• Avoid caching wide transformations unless memory is sufficient.
• Use unpersist() when done caching to free memory.

✅ Interview One-Liner:
Spark uses unified memory management where storage (cache) and execution share a pool. Proper partitioning and caching reduce recomputation.

📊 2. How to Monitor Memory & Disk in Spark UI

Access: http://<driver-host>:4040 (for active job) or Spark History Server for past jobs.

Key Tabs to Monitor:

Example View:

• Executors Tab shows:
  • Storage Memory Used: How much is used for cache
  • Disk spilled: How often memory overflowed
  • GC Time: High = memory pressure

Tip:

Use spark.eventLog.enabled = true to retain logs and view them via Spark History Server.

✅ Interview One-Liner:
The Executors tab is your best friend for spotting memory overflows, GC issues, or poor caching strategies.

⚖️ 3. Spark vs Hadoop Performance Benchmark (With Example)

Test: Word Count on 100 GB dataset (HDFS)

Hadoop MapReduce:

# Mapper → disk write → shuffle → reducer → disk write
Time Taken: ~40 mins
Disk I/O: 3x (intermediate + final)

PySpark:

rdd = spark.read.text("hdfs://data.txt")
counts = rdd.flatMap(lambda x: x.split()).map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y)
counts.write.text("hdfs://output")
# All in-memory except final write
Time Taken: ~4-6 mins

Benchmark Summary

✅ Interview One-Liner:
Spark achieves speedup via DAG optimization + memory computing, reducing disk I/O that slows down traditional Hadoop MapReduce.

📌 Summary: Memory and Performance Tuning in Practice

“How much memory for 500 GB input?” “When to choose 10 vs 50 executors?” Or even auto-tuning and adaptive query execution (AQE) walkthrough?

Perfect — let’s address all four questions with detailed breakdowns:

📦 1. “How Much Memory for 500 GB Input in Spark?”

🧮 Rule of Thumb for Memory Sizing:

Spark doesn’t load the entire 500 GB at once. It partitions it and processes it in chunks.

🔹 Considerations:

🧠 Example Estimation for 500 GB Parquet:

Assume:

• Snappy-compressed Parquet
• Effective size in RAM ≈ 1 TB
• Desired partition size = 128 MB → 8192 partitions (~1 TB / 128 MB)
• Running on 16-node cluster, each with 32 GB RAM and 8 cores

🔧 Suggested Config:

--executor-memory 8G
--executor-cores 4
--num-executors 40

Final Decision Guide:

✅ Interview Tip: “We size memory not just by input size, but by job behavior — wide transformations, skew, shuffle size — are more important.”

🎯 2. “When to Choose 10 vs 50 Executors?”

🔍 Executors are like distributed JVM workers. Choosing more/less depends on:

🔹 Considerations:

🔄 Examples:

• Join of 10M records + filtering → 10–15 executors
• Join of 1B+ rows (multiple tables) → 40–60 executors
• Streaming job → Keep executors minimal to avoid reallocation delay

🔧 Command line for 50 executors:

--executor-memory 6G --executor-cores 4 --num-executors 50

✅ Interview Tip: “I tune executor count based on shuffle stage parallelism and memory pressure. Bigger jobs with wide joins need more executors.”

🤖 3. Auto-Tuning & AQE (Adaptive Query Execution) — Spark 3+

✨ What is AQE?

AQE allows Spark to optimize the physical execution plan dynamically at runtime based on actual data statistics, not just estimates.

🔹 Key AQE Features:

🛠️ How to Enable:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

✅ Example:

If Spark sees one join side has < 10 MB at runtime, it automatically switches to broadcast join, even if plan had sort-merge initially.

📉 Without AQE: You have to guess join strategy
📈 With AQE: Spark picks best path based on live metrics

✅ Interview One-Liner:
“AQE gives Spark the ability to self-optimize join types, partitioning, and skew — critical in dynamic big data workloads.”

🧪 4. Benchmark Details for 100 GB Word Count (Hadoop vs Spark)

📌 Setup Recap:

🤝 Key Differences:

• Hadoop writes after map, before reduce: 2x disk I/O
• Spark pipes all transformations, keeps data in memory
• Spark reuses data in memory if cached (Hadoop cannot)

✅ Conclusion: Spark’s DAG-based, memory-first model yields massive speedups for iterative or multi-step transformations.

⚡ Summary Table: Spark Tuning Essentials

# 📘 Spark Tuning Guide Notebook — Prebuilt Reference

# ✅ Initialization (SparkSession + Configs)
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Spark Tuning Notebook") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.broadcastTimeout", "600") \
    .config("spark.sql.autoBroadcastJoinThreshold", 10485760) \
    .config("spark.sql.shuffle.partitions", 200) \
    .getOrCreate()

# 🔍 1. Check Join Strategy Logging
spark.conf.set("spark.sql.join.preferSortMergeJoin", "false")  # Prefer broadcast joins if possible
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 20 * 1024 * 1024)  # 20 MB

# Sample broadcast vs non-broadcast join
small_df = spark.read.parquet("/path/to/dim_table")  # ~10MB
large_df = spark.read.parquet("/path/to/fact_table")  # ~10GB

result = large_df.join(small_df.hint("broadcast"), "id").groupBy("region").count()

# 🧠 2. Repartitioning Strategies
from pyspark.sql.functions import col

# Default: no. of shuffle partitions (200)
df = spark.read.parquet("/path/large")
df = df.repartition(100, col("region"))  # Custom partitioning

# Coalesce for fewer files (e.g. writing to S3 or HDFS)
df = df.coalesce(10)
df.write.mode("overwrite").parquet("/out")

# 🧪 3. AQE Testing
# Run a query to test adaptive join switching
spark.sql("""
    SELECT c.country, COUNT(*)
    FROM big_sales s
    JOIN small_country c ON s.country_code = c.code
    GROUP BY c.country
""").show()

# 🔧 4. Cache vs Persist
cached = df.select("col1", "col2").cache()
cached.count()  # trigger caching

# To unpersist
cached.unpersist()

# 📉 5. Monitoring Memory + Join Plan
# View DAG and stages in Spark UI (port 4040)
df.explain(mode="formatted")  # View physical plan with broadcast info

# ✅ Tune this YAML config into Spark cluster
# 📄 spark-defaults.yaml

"""
spark.sql.adaptive.enabled true
spark.sql.adaptive.skewJoin.enabled true
spark.sql.adaptive.coalescePartitions.enabled true
spark.sql.autoBroadcastJoinThreshold 20971520
spark.sql.broadcastTimeout 600
spark.sql.shuffle.partitions 200
spark.sql.join.preferSortMergeJoin false
"""

# 🧾 Notes:
# - Adjust broadcast threshold depending on table size
# - Use coalesce before writing to prevent small file issues
# - Use explain() to confirm if broadcast joins are being triggered
# - AQE requires Spark 3.0+

Pages: 1 2 3