Shulou(Shulou.com)06/01 Report--

This article introduces you how to operate delta lake curd, the content is very detailed, interested friends can refer to, hope to be helpful to you.

Delta lake's tables support the syntax of deleting and updating data, so let's start with the syntax of sql and scala.

1. Delete delta data

You can delete data from the delta table according to the query conditions, such as deleting data with a date before 2017. The syntax for sql and scala is as follows.

Sql

DELETE FROM events WHERE date

< '2017-01-01' DELETE FROM delta.`/data/events/` WHERE date < '2017-01-01' scala import io.delta.tables._ val deltaTable = DeltaTable.forPath(spark, "/data/events/") deltaTable.delete("date < '2017-01-01'") // predicate using SQL formatted string import org.apache.spark.sql.functions._import spark.implicits._ deltaTable.delete(col("date") < "2017-01-01") // predicate using Spark SQL functions and implicits 请注意，delete操作会将数据从delta 表的最新版本中删除，但其实只有到历史版本直接被vacuum清空的时候，才会从物理存储中删除数据。 2. 更新表 可以更新满足条件的表。比如想更新eventType的字段字符串的编写失误，可以使用下面的表达，sql和scala的表达分别如下： sql UPDATE events SET eventType = 'click' WHERE eventType = 'clck'UPDATE delta.`/data/events/` SET eventType = 'click' WHERE eventType = 'clck' scala import io.delta.tables._ val deltaTable = DeltaTable.forPath(spark, "/data/events/") deltaTable.updateExpr( // predicate and update expressions using SQL formatted string "eventType = 'clck'", Map("eventType" ->

"'click'")

Import org.apache.spark.sql.functions._import spark.implicits._

DeltaTable.update (/ / predicate using Spark SQL functions and implicits col ("eventType") = = "clck", Map ("eventType"-> lit ("click")

3.merge operator to realize upsert Operation

Using the merge operation, you can upsert the data from the source table, view,dataframe to the target delta lake table. This operation is similar to traditional database merge into operations, but with additional support for delete operations, and additional conditions for updates, inserts, and deletes.

Suppose you generate a dataframe in the process of calculation, and the element is events, which contains eventId. And the eventId of part of the data in the dataframe already exists in the events table. At this point, you can use the merge into implementation, update its corresponding value if eventId exists, and insert its corresponding value if it does not exist. The implementation expression is as follows:

Sql

MERGE INTO eventsUSING updatesON events.eventId = updates.eventIdWHEN MATCHED THEN UPDATE SET events.data = updates.dataWHEN NOT MATCHED THEN INSERT (date, eventId, data) VALUES (date, eventId, data)

Scala

Import io.delta.tables._import org.apache.spark.sql.functions._

Val updatesDF =. / / define the updates DataFrame [date, eventId, data]

DeltaTable.forPath (spark, "/ data/events/") .as ("events") .merge (updatesDF.as ("updates"), "events.eventId = updates.eventId") .Map match .updateExpr (Map ("data"-> "updates.data")) .insertMatched .insertExpr (Map ("date"-> "updates.date", "eventId"-> "updates.eventId") "data"-> "updates.data"). Execute () curd operations on how to perform delta lake are shared here. I hope the above content can be of some help to you and learn more knowledge. If you think the article is good, you can share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.