Pyspark：使用转换类型作为字符串的多重连接条件

Posted 2023-04-17

【中文标题】Pyspark：使用转换类型作为字符串的多重连接条件

【英文标题】：Pyspark : Multiple join condition with cast type as string

【发布时间】：2021-01-18 02:47:48

【问题描述】：

在 Spark SQL 中，我需要将 as_of_date 转换为字符串并对 3 个表进行多重内连接，并在连接后选择 table1、2 和 3 中的所有行和列。示例表架构如下所示

Tablename : Table_01 alias t1

Column      | Datatype
as_of_date  | String
Tablename   | String
Credit_Card | String


Tablename : Table_02   alias t2

Column        | Datatype
as_of_date    | INT
Customer_name | String
tablename     | string

Tablename : Table_03   alias t3

Column        | Datatype
as_of_date    | String
tablename     | String 
address       | String

加入用例：

t1.as_of_date = t2.as_of_date AND t1.tablename = t2.tablename
t2.as_of_date = t3.as_of_date AND t2.tablename = t3.tablename      

   

表已在 hive 中创建，我正在对这些表进行 spark 转换，并将 table_02 中的 as_of_date 转换为字符串。

我想到了两种方法，但我不确定哪种方法最好

方法一：

df = spark.sql("select t1.*,t2.*,t3.* from table_1 t1 where cast(t1.as_of_date as string) inner join table_t2 t2 on t1.as_of_date = t2.as_of_date AND t1.tablename = t2.tablename inner join table_03 t3 on t2.as_of_date = t3.as_of_date and t2.tablename = t3.tablename")

方法二：

df_t1 = spark.sql("select * from table_01");
df_t2 = spark.sql("select * from table_02");
df_t3 = spark.sql("select * from table_03");

## Cast as_of_date as String if dtype as of date is int 
if dict(df_t2.dtypes)["as_of_date"] == 'int':
        df_t1["as_of_date"].cast(cast(StringType())
                                  
## Join Condition 
df = df_t1.alias('t1').join(df_t2.alias('t2'),on="t1.tablename=t2.tablename AND t1.as_of_date = t2.as_of_date", how='inner').join(df_t3.alias('t3'),on="t2.as_of_date = t3.as_of_date AND t2.tablename = t3.tablename",how='inner').select('t1.*,t2.*,t3.*')

我觉得使用方法 2 是冗长的，我需要一些关于我应该使用哪种方法的建议，以便于维护和使用的脚本

【参考方案1】：

我建议直接使用 Spark SQL，如下所示。无论数据类型如何，您都可以将所有表中的每个 as_of_date 列转换为字符串。您想将整数转换为字符串，但如果您还将字符串转换为字符串，则没有任何害处。

df = spark.sql("""
    select t1.*, t2.*, t3.*
    from t1
    join t2 on string(t1.as_of_date) = string(t2.as_of_date) AND t1.tablename = t2.tablename
    join t3 on string(t2.as_of_date) = string(t3.as_of_date) AND t2.tablename = t3.tablename
""")