Spark 2.x DataFrame值编码：如何为arcs_df映射nodes_df对应节点编号

Got it, let's work through this together. Since you’ve already added row numbers as unique node identifiers in your nodes_df, the solution boils down to using joins to map the original node values in arcs_df to their corresponding numeric IDs.

First, let’s clarify the assumed structure of your DataFrames (adjust column names to match your actual data):

• nodes_df: Contains your original node data plus the generated node_number (the row number you added). For example:

  original_node_name	node_number
  NodeA	1
  NodeB	2
  NodeC	3

• arcs_df: Contains edges with node_from and node_to referencing the original node values (like original_node_name above).

Step-by-Step Solution (Python) #

We’ll use aliases for nodes_df to avoid column name conflicts when joining twice (once for node_from, once for node_to):

# Import necessary modules (if not already done)
from pyspark.sql.functions import col

# Alias nodes_df to distinguish between "from" and "to" node lookups
from_nodes = nodes_df.alias("from_nodes")
to_nodes = nodes_df.alias("to_nodes")

# Perform joins to map original node values to their numeric IDs
result_df = arcs_df.join(
    # Join on node_from to get the corresponding node_number
    from_nodes,
    col("node_from") == from_nodes["original_node_name"],
    how="left"  # Use "inner" if you only want edges with valid nodes
).join(
    # Join on node_to to get the corresponding node_number
    to_nodes,
    col("node_to") == to_nodes["original_node_name"],
    how="left"
).select(
    # Rename the numeric IDs to match your desired output
    from_nodes["node_number"].alias("node_from"),
    to_nodes["node_number"].alias("node_to"),
    # Include any other columns from arcs_df you need to retain
    arcs_df["*"]
).drop(arcs_df["node_from"], arcs_df["node_to"])  # Remove the original text-based node columns

Step-by-Step Solution (Scala) #

If you’re using Scala, the logic is identical—just syntax changes:

// Alias nodes_df for dual joins
val fromNodes = nodes_df.as("from_nodes")
val toNodes = nodes_df.as("to_nodes")

val resultDf = arcs_df.join(
    fromNodes,
    arcs_df("node_from") === fromNodes("original_node_name"),
    "left"
).join(
    toNodes,
    arcs_df("node_to") === toNodes("original_node_name"),
    "left"
).select(
    fromNodes("node_number").alias("node_from"),
    toNodes("node_number").alias("node_to"),
    arcs_df("*")
).drop(arcs_df("node_from"), arcs_df("node_to"))

Key Notes to Keep in Mind #

• Join Type: Use how="left" if you want to retain all edges from arcs_df (even those with nodes not present in nodes_df—these will have null IDs). Use how="inner" if you only want edges where both nodes exist in nodes_df.
• Unique Node Identifiers: Ensure the column you’re joining on (like original_node_name) is unique in nodes_df. If there are duplicates, your join will create redundant rows—clean up duplicates in nodes_df first if needed.
• Row Number Consistency: Double-check that the node_number in nodes_df was generated with a stable partition/sort order (using row_number().over(Window.partitionBy(...).orderBy(...))). This ensures node IDs don’t change between runs.

内容的提问来源于stack exchange，提问作者Markus