apache-spark - 插入时在增量表中自动增加 id
问题描述
我有一个关于使用 pysparkSQL 与 delta 表合并 csv 文件的问题。我设法创建了 upsert 函数,如果匹配则更新,如果不匹配则插入。
我想将列添加ID
到最终的增量表中,并在每次插入数据时增加它。此列标识我们的增量表中的每一行。有没有办法把它到位?
def Merge(dict1, dict2):
res = {**dict1, **dict2}
return res
def create_default_values_dict(correspondance_df,marketplace):
dict_output = {}
for field in get_nan_keys_values(get_mapping_dict(correspondance_df, marketplace)):
dict_output[field] = 'null'
# We want to increment the id row each time we perform an insertion (TODO TODO TODO)
# if field == 'id':
# dict_output['id'] = col('id')+1
# else:
return dict_output
def create_matched_update_dict(mapping, products_table, updates_table):
output = {}
for k,v in mapping.items():
if k == 'source_name':
output['products.source_name'] = lit(v)
else:
output[products_table + '.' + k] = F.when(col(updates_table + '.' + v).isNull(), col(products_table + '.' + k)).when(col(updates_table + '.' + v).isNotNull(), col(updates_table + '.' + v))
return output
insert_dict = create_not_matched_insert_dict(mapping, 'products', 'updates')
default_dict = create_default_values_dict(correspondance_df_products, 'Cdiscount')
insert_values = Merge(insert_dict, default_dict)
update_values = create_matched_update_dict(mapping, 'products', 'updates')
delta_table_products.alias('products').merge(
updates_df_table.limit(20).alias('updates'),
"products.barcode_ean == updates.ean") \
.whenMatchedUpdate(set = update_values) \
.whenNotMatchedInsert(values = insert_values)\
.execute()
我试图增加id
函数中的列,create_default_values_dict
但它似乎不能正常工作,它不会自动增加 1。还有其他方法可以解决这个问题吗?提前致谢 :)
解决方案
Delta 不支持自增列类型。
一般来说,Spark 不使用自增 ID,而是倾向于单调递增的 ID。见functions.monotonically_increasing_id()
。
如果要实现自动增量行为,则必须使用多个 Delta 操作,例如,查询最大值 + 将其添加到row_number()
通过窗口函数计算的列中 + 然后写入。这是有问题的,原因有两个:
除非您引入外部锁定机制或其他方式来确保在找到最大值和写入之间不会发生对表的更新,否则您最终可能会得到无效数据。
使用
row_number()
会将并行度降低到 1,强制所有数据通过单个内核,这对于大数据将非常慢。
最重要的是,您真的不想在 Spark 中使用自增列。
希望这可以帮助。
推荐阅读
- vba - 将组合框添加到 userForm 时出现内存不足错误
- symfony - 如果在树枝中为空,则设置默认值
- python - 在 django rest 框架中的嵌套序列化程序中上传图像
- mybatis - 添加@Transactional函数解决游标继续打开后卡在批量插入
- pre-commit - 为我的存储库设置 Talisman 时提交更改时出现意外错误
- php - 如何按降序动态排序年份范围数组
- android-studio - 更新到新的 android studio artic fox (2020.3.1) 后出现主题问题
- symfony - 使用 botman 发送电子邮件
- javascript - 将大型 XML 文件读入 Javascript 的问题
- java - 如何从 CSV 等数据集中“垂直”读取