我在 MySQL 中有一个语句，我正在尝试将其转换为 PySpark:
my_table_name = default_engagements
UPDATE """ + my_table_name + """ SET Engagement = CASE WHEN LinkedAccountId IN ('123456778910', '1098765432123', '254325678912', '429576512356') THEN '808000000298' END WHERE Engagement IS NULL OR Engagement RLIKE '^[a-zA-Z]'; """
我在 python spark 中找到了这个例子:
from pyspark.sql import functions as F
update_func = (F.when(F.col('update_col') == replace_val, new_value)
.otherwise(F.col('update_col')))
但我不知道如何使它适应上面的 SQL。有人可以帮我语法吗？我想将新信息转换为 DF，以便将新 DF 写入 S3。
最佳答案
这是您要找的吗？
import pyspark.sql.functions as f
default_engagements = default_engagements.withColumn('Engagement', f.when((f.col('LinkedAccountId').isin('123456778910', '1098765432123', '254325678912', '429576512356'))&((f.col('Engagement').isNull())|(f.col('Engagement').rlike('^[a-zA-Z]'))), '808000000298').otherwise(f.col('Engagement'))
关于mysql - 将 SQL 语句转换为 PySpark，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/58066776/