如何使用數(shù)據(jù)版本控制管理數(shù)據(jù)湖中的模式驗證
譯文譯者 | 李睿
審校 | 重樓
數(shù)據(jù)團隊依賴許多其他“第三方”發(fā)送數(shù)據(jù)的情況并不少見，他們經(jīng)常在沒有進行任何溝通或讓數(shù)據(jù)團隊知道太晚的情況下更改數(shù)據(jù)的模式。
每當(dāng)發(fā)生這種情況時，數(shù)據(jù)管道就會遭到破壞，數(shù)據(jù)團隊需要修復(fù)數(shù)據(jù)湖。這是一個充滿繁重任務(wù)的人工過程。在通常情況下，數(shù)據(jù)團隊可能會推卸責(zé)任，試圖證明模式已經(jīng)改變。
但是隨著發(fā)展和進步，數(shù)據(jù)團隊意識到，以自動持續(xù)集成（CI）/持續(xù)交付（CD）的方式簡單地阻止模式一起更改是更明智的。
模式更改和模式驗證給數(shù)據(jù)團隊帶來了很多痛苦，但是市場上有一些解決方案可以幫助解決這個問題——幸運的是，其中一些是開源的。
以下是一個循序漸進的教程，介紹如何使用開源數(shù)據(jù)版本控制工具lakeFS解決模式驗證問題。
什么是模式驗證?
模式驗證允許用戶為數(shù)據(jù)湖創(chuàng)建驗證規(guī)則，例如允許的數(shù)據(jù)類型和值范圍。它保證保存在數(shù)據(jù)湖中的數(shù)據(jù)遵循已建立的模式，該模式描述了數(shù)據(jù)的結(jié)構(gòu)、格式和限制。
由于用戶的數(shù)據(jù)湖可以填充來自具有不同模式定義的各種來源的數(shù)據(jù)，因此在數(shù)據(jù)湖中的所有數(shù)據(jù)上強制使用統(tǒng)一的模式是一個挑戰(zhàn)。
這是一個需要解決的問題——如果不快速采取行動，就會在數(shù)據(jù)處理過程中看到不一致和錯誤。
為什么需要處理模式驗證?
花費一些時間正確地管理模式是值得的，有以下四個原因：
- 一致性——數(shù)據(jù)湖通常包含來自多個來源的大量數(shù)據(jù)。如果沒有模式驗證，最終可能會以不一致或不正確的形式存儲在數(shù)據(jù)湖中，從而導(dǎo)致處理過程中的問題。
- 質(zhì)量——模式驗證通過施加數(shù)據(jù)限制和標(biāo)準(zhǔn)，有助于保持?jǐn)?shù)據(jù)湖中數(shù)據(jù)的良好質(zhì)量。它可以幫助用戶識別和標(biāo)記數(shù)據(jù)質(zhì)量問題，例如丟失或不準(zhǔn)確的信息，在它們導(dǎo)致下游出現(xiàn)問題之前。
- 效率——模式驗證通過確保數(shù)據(jù)湖中所有數(shù)據(jù)的統(tǒng)一模式來加快數(shù)據(jù)處理和分析。這反過來又減少了清理、轉(zhuǎn)換和分析數(shù)據(jù)所需的時間和精力，并提高了數(shù)據(jù)管道的總體效率。
- 合規(guī)性——許多企業(yè)必須滿足嚴(yán)格的監(jiān)管和合規(guī)性要求。模式驗證有助于確保存儲在數(shù)據(jù)湖中的數(shù)據(jù)符合這些標(biāo)準(zhǔn)，從而提供對數(shù)據(jù)沿襲和質(zhì)量的清晰審計跟蹤。
處理數(shù)據(jù)湖中的模式并非一帆風(fēng)順
在數(shù)據(jù)倉庫中，用戶處理的是嚴(yán)格的數(shù)據(jù)模型和嚴(yán)格的模式。數(shù)據(jù)湖與之相反。大多數(shù)情況下，它們最終包含廣泛的數(shù)據(jù)源。
為什么這很重要?因為在數(shù)據(jù)湖中，模式的定義可以在數(shù)據(jù)源之間發(fā)生變化，并且當(dāng)添加新數(shù)據(jù)時，模式可能會隨著時間的推移而變化。這使得在數(shù)據(jù)湖中的所有數(shù)據(jù)上實施統(tǒng)一的模式成為一個巨大的挑戰(zhàn)。如果不能解決這個問題，將不得不解決數(shù)據(jù)處理問題。
但這還不是全部。由于構(gòu)建在數(shù)據(jù)湖之上的數(shù)據(jù)管道的復(fù)雜性不斷增加，無法擁有一個一致的模式。數(shù)據(jù)管道可以包括多個流程和轉(zhuǎn)換，每個流程和轉(zhuǎn)換都需要一個唯一的模式定義。
模式可能隨著數(shù)據(jù)的處理和修改而變化，因此很難確保跨整個管道進行模式驗證。
這就是版本控制系統(tǒng)可以派上用場的地方。
在數(shù)據(jù)湖中實現(xiàn)模式驗證的數(shù)據(jù)版本控制
lakeFS是一個開源工具，它可以將數(shù)據(jù)湖轉(zhuǎn)換為類似Git的存儲庫，讓用戶像軟件工程師管理代碼一樣管理它。這就是數(shù)據(jù)版本控制的意義所在。
與其他源代碼控制系統(tǒng)一樣，lakeFS有一個稱為hook的特性，它是定制的腳本或程序，lakeFS平臺可以運行這些腳本或程序來響應(yīng)指定的事件或操作。
這些事件可以包括提交更改、合并分支、創(chuàng)建新分支、添加或刪除標(biāo)記等等。例如，當(dāng)合并發(fā)生時，在合并完成之前，在源分支上運行一個預(yù)合并掛鉤。
它如何應(yīng)用于模式驗證呢? 用戶可以創(chuàng)建一個預(yù)合并掛鉤來驗證Parquet文件的模式與當(dāng)前模式是否相同。
需要準(zhǔn)備什么
- lakeFS服務(wù)器(可以免費安裝或在云中啟動)。
- 可選：可以使用sample-repo來啟動一個筆記本（notebook），筆記本可以配置為連接到lakeFS服務(wù)器。
在這個場景中，將在一個攝取分支中創(chuàng)建一個delta表，并將其合并到生產(chǎn)中。接下來將更改表的模式，并嘗試再次合并它，模擬將數(shù)據(jù)提升到生產(chǎn)的過程。
1.設(shè)置
首先，將設(shè)置一些全局變量并安裝將在本例中使用的包，這些包將在Python筆記本中運行。
在設(shè)置好lakeFS憑證后，可以開始創(chuàng)建一些包含存儲庫和分支名稱的全局變量：
Python
repo = "schema-validation-example-repo"
mainBranch = "main"
ingestionBranch = "ingestion_branch"每個lakeFS存儲庫都需要有自己的存儲命名空間，所以也需要創(chuàng)建一個：
Python
storageNamespace = 's3://' # e.g. "s3://username-lakefs-cloud/"在本例中，使用AWS S3存儲。為了使一切順利進行，用戶的存儲需要配置為與lakeFS一起運行，lakeFS與AWS、Azure、Google Cloud或內(nèi)部部署對象存儲(如MinIO)一起工作。
如果在云中運行l(wèi)akeFS，則可以通過復(fù)制示例存儲庫的存儲名稱空間并將字符串附加到其上，將其鏈接到存儲。所以，如果lakeFS Cloud提供了這個sample-repo：
可以通過以下方式進行配置：
Python
storageNamespace = 's3://lakefs-sample-us-east-1-production/AROA5OU4KHZHHFCX4PTOM:2ae87b7718e5bb16573c021e542dd0ec429b7ccc1a4f9d0e3f17d6ee99253655/my_random_string'在筆記本中，將使用Python代碼，因此也必須導(dǎo)入lakeFS Python客戶端包：
Python
import lakefs_client
from lakefs_client import models
from lakefs_client.client import LakeFSClient
import os
from pyspark.sql.types import ByteType, IntegerType, LongType, StringType, StructType, StructField接下來，配置客戶端：
Python
%xmode Minimal
if not 'client' in locals():
# lakeFS credentials and endpoint
configuration = lakefs_client.Configuration()
configuration.username = lakefsAccessKey
configuration.password = lakefsSecretKey
configuration.host = lakefsEndPoint
client = LakeFSClient(configuration)
print("Created lakeFS client.")以下將在本例中創(chuàng)建delta表，因此需要包括以下包：
Python
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages io.delta:delta-core_2.12:2.0.0 --conf "spark.sql.extensinotallow=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" pyspark-shell'lakeFS公開了一個S3網(wǎng)關(guān)，它允許應(yīng)用程序以與S3通信的方式與lakeFS進行接口。要配置網(wǎng)關(guān)，并執(zhí)行以下步驟：
Python
from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
sc = SparkContext.getOrCreate()
spark = SparkSession(sc)
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", lakefsAccessKey)
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", lakefsSecretKey)
sc._jsc.hadoopConfiguration().set("fs.s3a.endpoint", lakefsEndPoint)
sc._jsc.hadoopConfiguration().set("fs.s3a.path.style.access", "true")現(xiàn)在已經(jīng)準(zhǔn)備好在筆記本中大規(guī)模使用lakeFS版本控制。
2.創(chuàng)建存儲庫和掛鉤
以下將使用Python客戶端創(chuàng)建存儲庫：
Python
client.repositories.create_repository(
repository_creatinotallow=models.RepositoryCreation(
name=repo,
storage_namespace=storageNamespace,
default_branch=mainBranch))在這種情況下，將使用預(yù)合并掛鉤來確保架構(gòu)沒有更改。操作文件應(yīng)提交到lakeFS存儲庫，前綴為_lakeFS_actions/。未能分析操作文件將導(dǎo)致運行失敗。
將提交以下鉤子配置操作文件，pre-merge-schema-validation.yaml：
Python
#Parquet schema Validator
#Args:
# - locations (list of strings): locations to look for parquet files under
# - sample (boolean): whether reading one new/changed file per directory is enough, or go through all of them
#Example hook declaration: (_lakefs_actions/pre-merge-schema-validation.yaml):
name: pre merge checks on main branch
on:
、
pre-merge:
branches:
- main
hooks:
- id: check_schema_changes
type: lua
properties:
script_path: scripts/parquet_schema_change.lua # location of this script in the repository
args:
sample: false
locations:
- tables/customers/這個文件（pre-merge-schema-validation.yaml）存儲在example repo中的子文件夾LuaHooks中。必須將文件提交到文件夾_lakeFS_actions下的lakeFS存儲庫：
Python
hooks_config_yaml = "pre-merge-schema-validation.yaml"
hooks_prefix = "_lakefs_actions"
with open(f'./LuaHooks/{hooks_config_yaml}', 'rb') as f:
client.objects.upload_object(repository=repo,
branch=mainBranch,
path=f'{hooks_prefix}/{hooks_config_yaml}',
cnotallow=f
)只是設(shè)置了一個動作腳本，在合并到main之前運行scripts/parquet_schema_che.lua。
然后將創(chuàng)建腳本本身（parquet_schema_che.lua）并將其上載到腳本目錄中。正如人們所看到的，使用嵌入式LuaVM來運行鉤子，而不依賴于其他組件。
此文件也位于ample-repo中的LuaHooks子文件夾中：
Python
--[[
Parquet schema validator
Args:
- locations (list of strings): locations to look for parquet files under
- sample (boolean): whether reading one new/changed file per directory is enough, or go through all of them
]]
lakefs = require("lakefs")
strings = require("strings")
parquet = require("encoding/parquet")
regexp = require("regexp")
path = require("path")
visited_directories = {}
for _, location in ipairs(args.locations) do
after = ""
has_more = true
need_more = true
print("checking location: " .. location)
while has_more do
print("running diff, location = " .. location .. " after = " .. after)
local code, resp = lakefs.diff_refs(action.repository_id, action.branch_id, action.source_ref, after, location)
if code ~= 200 then
error("could not diff: " .. resp.message)
end
for _, result in pairs(resp.results) do
p = path.parse(result.path)
print("checking: '" .. result.path .. "'")
if not args.sample or (p.parent and not visited_directories[p.parent]) then
if result.path_type == "object" and result.type ~= "removed" then
if strings.has_suffix(p.base_name, ".parquet") then
-- check it!
code, content = lakefs.get_object(action.repository_id, action.source_ref, result.path)
if code ~= 200 then
error("could not fetch data file: HTTP " .. tostring(code) .. "body:\n" .. content)
end
schema = parquet.get_schema(content)
for _, column in ipairs(schema) do
for _, pattern in ipairs(args.column_block_list) do
if regexp.match(pattern, column.name) then
error("Column is not allowed: '" .. column.name .. "': type: " .. column.type .. " in path: " .. result.path)
end
end
end
print("\t all columns are valid")
visited_directories[p.parent] = true
end
end
else
print("\t skipping path, directory already sampled")
end
end
-- pagination
has_more = resp.pagination.has_more
after = resp.pagination.next_offset
end
end把文件（這次是parquet_schema_che.lua）從LuaHooks目錄上傳到lakeFS存儲庫中操作配置文件中指定的位置（即腳本文件夾內(nèi)）：
Python
hooks_config_yaml = "pre-merge-schema-validation.yaml"
hooks_prefix = "_lakefs_actions"
with open(f'./LuaHooks/{hooks_config_yaml}', 'rb') as f:
client.objects.upload_object(repository=repo,
branch=mainBranch,
path=f'{hooks_prefix}/{hooks_config_yaml}',
content=f
)必須在提交操作文件后提交更改才能生效：
Python
client.commits.commit(
repository=repo,
branch=mainBranch,
commit_creatinotallow=models.CommitCreation(
message='Added hook config file and schema validation scripts'))如果切換到lakeFS UI，應(yīng)該會在主目錄下看到以下目錄結(jié)構(gòu)和文件：
LakeFS UI的目錄結(jié)構(gòu)
lakeFS UI中顯示的合并前架構(gòu)驗證
lakeFS UI中的架構(gòu)驗證腳本
3.使用原始模式運行第一個ETL
在lakeFS中，可以在與生產(chǎn)（主要）分支不同的分支上進行攝取和轉(zhuǎn)化。
以下將建立一個攝取分支：
Python
client.branches.create_branch(
repository=repo,
branch_creatinotallow=models.BranchCreation(
name=ingestionBranch, source=mainBranch))接下來，將使用Kaggle數(shù)據(jù)集Orion Star——運動和戶外RDBMS數(shù)據(jù)集。使用Customer.csv，可以從data/samples/OrionStar/將其上傳到示例存儲庫。
首先，需要定義表模式：
Python
customersSchema = StructType([
StructField("User_ID", IntegerType(), False),
StructField("Country", StringType(), False),
StructField("Gender", StringType(), False),
StructField("Personal_ID", IntegerType(), True),
StructField("Customer_Name", StringType(), False),
StructField("Customer_FirstName", StringType(), False),
StructField("Customer_LastName", StringType(), False),
StructField("Birth_Date", StringType(), False),
StructField("Customer_Address", StringType(), False),
StructField("Street_ID", LongType(), False),
StructField("Street_Number", IntegerType(), False),
StructField("Customer_Type_ID", IntegerType(), False)
])然后，從CSV文件中，將創(chuàng)建一個delta表，并將其提交到存儲庫：
Python
customersTablePath = f"s3a://{repo}/{ingestionBranch}/tables/customers"
df = spark.read.csv('./data/samples/OrionStar/CUSTOMER.csv',header=True,schema=customersSchema)
df.write.format("delta").mode("overwrite").save(customersTablePath)在這里需要做出改變：
Python
client.commits.commit(
repository=repo,
branch=ingestionBranch,
commit_creatinotallow=models.CommitCreation(
message='Added customers Delta table',
metadata={'using': 'python_api'}))然后，使用合并將數(shù)據(jù)發(fā)送到生產(chǎn)：
Python
client.refs.merge_into_branch(
repository=repo,
source_ref=ingestionBranch,
destination_branch=mainBranch)已經(jīng)完成的架構(gòu)驗證序列：
4. 修改模式并嘗試將表移動到生產(chǎn)環(huán)境
為了簡化操作，將重命名其中一列。以下將Country_name替換為Country_name：
Python
customersSchema = StructType([
StructField("User_ID", IntegerType(), False),
StructField("Country_Name", StringType(), False), # Column name changes from Country to Country_name
StructField("Gender", StringType(), False),
StructField("Personal_ID", IntegerType(), True),
StructField("Customer_Name", StringType(), False),
StructField("Customer_FirstName", StringType(), False),
StructField("Customer_LastName", StringType(), False),
StructField("Birth_Date", StringType(), False),
StructField("Customer_Address", StringType(), False),
StructField("Street_ID", LongType(), False),
StructField("Street_Number", IntegerType(), False),
StructField("Customer_Type_ID", IntegerType(), False)
])在攝取分支中，重新創(chuàng)建delta表：
Python
customersTablePath = f"s3a://{repo}/{ingestionBranch}/tables/customers"
df = spark.read.csv('./data/samples/OrionStar/CUSTOMER.csv',header=True,schema=customersSchema)
df.write.format("delta").mode("overwrite").option("overwriteSchema", "true").save(customersTablePath)
])需要進行修改：
Python
client.commits.commit(
repository=repo,
branch=ingestionBranch,
commit_creatinotallow=models.CommitCreation(
message='Added customers table with schema changes',
metadata={'using': 'python_api'}))然后，可以嘗試將數(shù)據(jù)投入生產(chǎn)：
Python
client.commits.commit(
repository=repo,
branch=ingestionBranch,
commit_creatinotallow=models.CommitCreation(
message='Added customer tables with schema changes!',
metadata={'using': 'python_api'}))由于模式修改，得到了一個先決條件Failed錯誤。合并前的掛鉤阻礙了晉升。因此，這些數(shù)據(jù)不會在生產(chǎn)中使用：
從lakeFS UI中，可以導(dǎo)航到存儲庫并選擇“Actions”選項。接下來，單擊失敗操作的Run ID，選擇“主分支上的合并前檢查”，展開check_schema_changes，并查看錯誤消息。
結(jié)語
由于存儲數(shù)據(jù)的異構(gòu)性和原始性，數(shù)據(jù)湖上的模式驗證至關(guān)重要，但也很困難。管理模式演變、數(shù)據(jù)轉(zhuǎn)換和跨多種格式的兼容性檢查意味著每個數(shù)據(jù)從業(yè)者都需要一些非常強大的方法和工具。
數(shù)據(jù)湖的去中心化性質(zhì)，許多用戶和系統(tǒng)可以在其中編輯數(shù)據(jù)，使模式驗證更加復(fù)雜。模式的驗證對于數(shù)據(jù)治理、集成和可靠的分析至關(guān)重要。
像上面展示的預(yù)合并掛鉤這樣的解決方案有助于在將模式文件合并到生產(chǎn)分支之前驗證它們。它在保證數(shù)據(jù)完整性和防止不兼容的模式更改合并到主分支時非常方便。它還增加了一層額外的質(zhì)量控制，使數(shù)據(jù)更加一致。
原文標(biāo)題：Managing Schema Validation in a Data Lake Using Daa Version Control，作者：Iddo Avneri
