Táblaséma frissítése sémafejlődéssel

A táblák támogatják a sémafejlődést, lehetővé téve a táblaszerkezet módosítását az adatkövetelmények változásakor. A következő típusváltozások támogatottak:

Végezze el ezeket a módosításokat kifejezetten DDL használatával vagy implicit módon a DML használatával.

Important

A sémafrissítések ütköznek az összes egyidejű írási művelettel. A Databricks a sémamódosítások összehangolását javasolja az írási ütközések elkerülése érdekében.

A táblaséma frissítése leállítja a táblából beolvasott adatfolyamokat. A feldolgozás folytatásához indítsa újra a streamet a strukturált stream éles környezetben ismertetett módszereivel.

Manuális sémamódosítások

Utasítások használatával ALTER TABLE explicit módon módosíthatja a tábla sémáját új adatok írása nélkül.

Oszlopok hozzáadása

Egy vagy több oszlop meglévő táblához való hozzáadására használható ALTER TABLE ... ADD COLUMNS , opcionálisan pozíció és megjegyzés megadásával:

ALTER TABLE table_name ADD COLUMNS (col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Alapértelmezés szerint a nullabilitás a következő true: .

Példa: Beágyazott mezők hozzáadása

Beágyazott oszlopok hozzáadása csak a szerkezetek esetében támogatott. A tömbök és a térképek nem támogatottak.

Ha oszlopot szeretne hozzáadni egy beágyazott mezőhöz, használja a következőt:

ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

Ha például a séma futtatása ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1) előtt a következő:

- root
| - colA
| - colB
| +-field1
| +-field2

a séma a következő után következik:

- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2

Oszlopmegjegyzések és sorrend módosítása

Egy oszlop megjegyzésének frissítésére vagy átrendezésére használható ALTER TABLE ... ALTER COLUMN más oszlopokhoz képest:

ALTER TABLE table_name ALTER [COLUMN] col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Példa: Beágyazott mezők módosítása

Beágyazott mező oszlopának módosításához használja a következőt:

ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name (COMMENT col_comment | FIRST | AFTER colA_name)

Ha például a séma futtatása ALTER TABLE boxes ALTER COLUMN colB.field2 FIRST előtt a következő:

- root
| - colA
| - colB
| +-field1
| +-field2

a séma a következő után következik:

- root
| - colA
| - colB
| +-field2
| +-field1

Cserélje le az oszlopokat

A ALTER TABLE ... REPLACE COLUMNS tábla teljes oszloplistájának újradefiniálása, beleértve az oszlopok hozzáadását, eltávolítását, átrendezését vagy átnevezését egyetlen műveletben:

ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)

Példa: Beágyazott mezők cseréje

Például a következő DDL futtatásakor:

ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)

ha az előző séma a következő:

- root
| - colA
| - colB
| +-field1
| +-field2

a séma a következő után következik:

- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA

Oszlopok átnevezése

Az oszlopok átnevezéséhez az oszlopok meglévő adatainak újraírása nélkül engedélyeznie kell az oszlopleképezést a táblához. Lásd: Oszlopok átnevezése és elhagyása a Delta Lake oszlopleképezés alkalmazásával.

Az oszlop átnevezése:

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

Példa: Beágyazott mezők átnevezése

Beágyazott mező átnevezése:

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

Ha például a következő parancsot futtatja:

ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001

Ha az előző séma a következő:

- root
| - colA
| - colB
| +-field1
| +-field2

Ezután a séma a következő:

- root
| - colA
| - colB
| +-field001
| +-field2

Lásd: Oszlopok átnevezése és elhagyása a Delta Lake oszlopleképezés alkalmazásával.

Oszlopok törlése

Ha csak metaadat-műveletként szeretné elvetni az oszlopokat adatfájlok újraírása nélkül, engedélyeznie kell a tábla oszlopleképezését. Lásd: Oszlopok átnevezése és elhagyása a Delta Lake oszlopleképezés alkalmazásával.

Note

Az oszlop metaadatokból való elvetése nem törli a fájlokban lévő oszlop alapjául szolgáló adatokat. A törölt oszlop adatainak törléséhez:

  • Fájlok átírására használható REORG TABLE .
  • Ezután a VACUUM használatával fizikailag törölheti az eltávolított oszlop adatait tartalmazó fájlokat.

Oszlop elvetése:

ALTER TABLE table_name DROP COLUMN col_name

Több oszlop eltávolítása:

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

Oszloptípus vagy -név módosítása

Módosíthatja egy oszlop típusát vagy nevét, vagy elvethet egy oszlopot a tábla újraírásával. Ehhez használja a overwriteSchema lehetőséget.

Az alábbi példa egy oszloptípus módosítását mutatja be:

(spark.read.table(...)
  .withColumn("birthDate", col("birthDate").cast("date"))
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

Az alábbi példa egy oszlopnév módosítását mutatja be:

(spark.read.table(...)
  .withColumnRenamed("dateOfBirth", "birthDate")
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

Sémafejlődés engedélyezése

A meglévő táblába WITH SCHEMA EVOLUTION vagy mergeSchema kívánt adatok sémája alapján történő sémamódosítások elvégzéséhez használja a(z) true elemet, vagy állítsa a(z) INSERT értékét MERGE értékre.

A sémafejlődés engedélyezése az alábbi módszerek egyikével:

A Databricks azt javasolja, hogy a Spark-konfiguráció beállítása helyett engedélyezze a sémafejlődést minden írási művelethez a WITH SCHEMA EVOLUTION szintaxis vagy a mergeSchema beállítás használatával.

Ha beállításokat vagy szintaxist használ a sémafejlődés engedélyezéséhez egy írási műveletben, ez elsőbbséget élvez a Spark-konfigurációval szemben.

Sémafejlődés engedélyezése írásokhoz új oszlopok hozzáadásához

Ha a sémafejlődés engedélyezve van, a forrás lekérdezésben található, de a céltáblából hiányzó oszlopok automatikusan hozzáadódnak egy írási tranzakció részeként. Lásd: Sémafejlődés engedélyezése.

Vegye figyelembe a következőket:

  • Az eredeti kis- és nagybetűk megmaradnak egy új oszlop hozzáadása során.
  • A rendszer új oszlopokat ad hozzá a táblaséma végéhez.
  • Ha a további oszlopok egy struktúra részét képezik, a rendszer hozzáfűzi őket a céltáblában lévő struktúra végéhez.

INSERT SQL használatával végzett sémaevolúcióval

A Databricks Runtime 18.1-ben és újabb verziókban használja az utasításokban szereplő WITH SCHEMA EVOLUTION záradékot a INSERT sémafejlődés engedélyezéséhez:

INSERT WITH SCHEMA EVOLUTION INTO target_table
SELECT * FROM source_table

Ha a lekérdezés source_table olyan oszlopokat ad vissza, amelyek nem szerepelnek a céltáblában, a rendszer automatikusan hozzáadja ezeket az oszlopokat a target_table sémához. A meglévő sorok értékeket kapnak NULL az új oszlopokhoz.

A WITH SCHEMA EVOLUTION záradék támogatja a , INSERT INTOés INSERT OVERWRITE az INSERT INTO ... REPLACEűrlapokat. A célnak Delta Lake- vagy Apache Iceberg-táblának kell lennie. Ha ezzel a klauzulával szúr be adatokat egy Hive-táblába vagy más nem-Delta táblába, az hibát eredményez.

Ha a Databricks Runtime 18.0-s vagy újabb verziót használja, engedélyezze inkább a sémafejlődést a mergeSchema beállítással. Tekintse meg INSERT a sémafejlődést a DataFrame API használatával.

INSERT sémafejlődés a DataFrame API használatával

Az alábbi példa azt mutatja be, hogy a mergeSchema lehetőséget kötegírási művelettel használja:

Python
(spark.read
  .table("source_table")
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("target_table")
)
Scala
spark.read
  .table("source_table")
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("target_table")

INSERT sémafejlődés strukturált streameléssel

Az alábbi példa bemutatja a mergeSchema beállítás használatát az Auto Loaderrel a Structured Streamingben. Lásd : Mi az automatikus betöltő?.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

Automatikus sémafejlődés az egyesítéshez

A MERGEsémafejlődés lehetővé teszi a sémaeltérések feloldását a cél és a forrástábla között. A következő két esetet kezeli:

  1. A forrástáblában egy oszlop található, a céltáblában nem, és név alapján van megadva a beszúrási vagy frissítési műveletek hozzárendelésében. Másik lehetőségként egy UPDATE SET * vagy INSERT * művelet van jelen.

    Ez az oszlop hozzá lesz adva a célsémához, és az értékei a forrás megfelelő oszlopából lesznek feltöltve.

    • Ez csak akkor érvényes, ha az egyesítési forrás oszlopneve és struktúrája pontosan egyezik a célhozzárendeléssel.

    • Az új oszlopnak szerepelnie kell a forrássémában. Az új oszlop hozzárendelése a műveleti záradékban nem határozza meg ezt az oszlopot.

    Ezek a példák lehetővé teszik a séma fejlődését:

    -- The column newcol is present in the source but not in the target. It will be added to the target.
UPDATE SET target.newcol = source.newcol

-- The field newfield doesn't exist in struct column somestruct of the target. It will be added to that struct column.
UPDATE SET target.somestruct.newfield = source.somestruct.newfield

-- The column newcol is present in the source but not in the target.
-- It will be added to the target.
UPDATE SET target.newcol = source.newcol + 1

-- Any columns and nested fields in the source that don't exist in target will be added to the target.
UPDATE SET *
INSERT *

    Ezek a példák nem aktiválják a sémafejlődést, ha az oszlop newcol nem szerepel a source sémában:

    UPDATE SET target.newcol = source.someothercol
UPDATE SET target.newcol = source.x + source.y
UPDATE SET target.newcol = source.output.newcol

  2. A céltáblában van egy oszlop, a forrástáblában azonban nem.

    A célséma nem módosul. Ezek az oszlopok:

    • A rendszer változatlanul hagyja a következőt: UPDATE SET *.

    • A NULLINSERT * értékre van beállítva.

    • A művelet záradékban való hozzárendelés esetén a program továbbra is explicit módon módosíthatja azokat.

    Például:

    UPDATE SET *  -- The target columns that are not in the source are left unchanged.
INSERT *  -- The target columns that are not in the source are set to NULL.
UPDATE SET target.onlyintarget = 5  -- The target column is explicitly updated.
UPDATE SET target.onlyintarget = source.someothercol  -- The target column is explicitly updated from some other source column.

Manuálisan kell engedélyeznie az automatikus sémafejlődést. Lásd: Sémafejlődés engedélyezése.

Note

A Databricks Runtime 11.3 LTS-ben és alatta csak INSERT * vagy UPDATE SET * műveletek használhatók a sémafejlődéshez az egyesítéssel.

A Databricks Runtime 12.2 LTS-ben és újabb verziókban a forrástáblában található oszlopok és strukturált mezők név szerint adhatók meg a beszúrási vagy frissítési műveletekben.

A Databricks Runtime 13.3 LTS-ben és újabb verziókban sémafejlődést használhat a térképekbe ágyazott szerkezetekkel, például map<int, struct<a: int, b: int>>.

MERGEsémafejlődés sql, Python és Scala használatával

A Databricks Runtime 15.4 LTS-ben és újabb verziókban sql- vagy tábla API-k használatával megadhatja a sémafejlődést egy egyesítési utasításban:

SQL
MERGE WITH SCHEMA EVOLUTION INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE
Python
from delta.tables import *

(targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)
Scala
import io.delta.tables._

targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

Példaműveletek sémafejlődéssel MERGE

Íme néhány példa a sémafejlődéssel és anélkül végzett működés hatásaira MERGE .

Columns Lekérdezés (SQL-ben) Viselkedés sémafejlődés nélkül (alapértelmezett) Viselkedés sémafejlődéssel
Céloszlopok: key, value
Forrásoszlopok: key, value, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 A táblaséma marad változatlan; csak az oszlopok key, value frissülnek/beszúródnak. A táblaséma a következőre (key, value, new_value)módosul: . A meglévő egyezésekkel rendelkező rekordok frissülnek a forrásban lévő value és new_value értékekkel. A rendszer új sorokat szúr be a sémával (key, value, new_value).
Céloszlopok: key, old_value
Forrásoszlopok: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET *
WHEN NOT MATCHED
THEN INSERT *		 UPDATE és INSERT a műveletek hibát jeleznek, mert a céloszlop old_value nem szerepel a forrásban. A táblaséma a következőre (key, old_value, new_value)módosul: . A meglévő egyezésekkel rendelkező rekordok frissülnek a new_value forrásban változatlanul hagyva old_value . A rendszer új rekordokat szúr be a megadott key, new_value és NULL a old_value számára.
Céloszlopok: key, old_value
Forrásoszlopok: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED
THEN UPDATE SET new_value = s.new_value		 UPDATE hibát jelez, mert az oszlop new_value nem létezik a céltáblában. A táblaséma a következőre (key, old_value, new_value)módosul: . A meglévő egyezésekkel rendelkező rekordok frissülnek, a new_value kerül be a forrásba, míg a old_value változatlan marad, és a nem egyező rekordoknál a NULL kerül be a new_value mezőbe. Lásd az 1. megjegyzést.
Céloszlopok: key, old_value
Forrásoszlopok: key, new_value		 MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN NOT MATCHED
THEN INSERT (key, new_value) VALUES (s.key, s.new_value)		 INSERT hibát jelez, mert az oszlop new_value nem létezik a céltáblában. A táblaséma a következőre (key, old_value, new_value)módosul: . A rendszer új rekordokat szúr be a megadott key, new_value és NULL a old_value számára. Létező rekordok esetében a NULL mezőbe beírt adatok megmaradnak, míg a new_value változatlan marad. Lásd az 1. megjegyzést.

(1) Ez a viselkedés a Databricks Runtime 12.2 LTS és az újabb verziókban érhető el; ebben a feltételben a Databricks Runtime 11.3 LTS és az ennél korábbi verziók hibát jeleznek.

Oszlopok kizárása egyesítéssel

A Databricks Runtime 12.2 LTS és újabb verzióiban az egyesítési feltételekben záradékokat használhat az oszlopok explicit kizárásához. A kulcsszó viselkedése EXCEPT attól függően változik, hogy engedélyezve van-e a sémafejlődés.

Ha a sémafejlődés ki van kapcsolva, a EXCEPT kulcsszó a céltábla oszlopainak listájára vonatkozik, és lehetővé teszi, hogy oszlopokat zárjanak ki a UPDATE vagy INSERT műveletekből. A kizárt oszlopok értéke null lesz beállítva.

Ha a sémafejlődés engedélyezve van, a EXCEPT kulcsszó a forrástáblában lévő oszlopok listájára vonatkozik, és lehetővé teszi az oszlopok kizárását a sémafejlődésből. A forrás új oszlopa, amely nem szerepel a céltáblában, nem lesz hozzáadva a célsémához, ha szerepel a EXCEPT záradékban. A célban már meglévő kizárt oszlopok a következőre nullvannak állítva: .

Példák a(z) EXCLUDE és a(z) MERGE együttes használatára

Az alábbi példák ezt a szintaxist mutatják be:

Columns Lekérdezés (SQL-ben) Viselkedés sémafejlődés nélkül (alapértelmezett) Viselkedés sémafejlődéssel
Céloszlopok: id, title, last_updated
Forrásoszlopok: id, title, review, last_updated		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated)		 A egyeztetett sorok úgy frissülnek, hogy a last_updated mezőt az aktuális dátumra állítja. Az új sorok beszúrása a következő értékekkel id történik: és title. A kizárt mező last_updated értéke : null. A mező review figyelmen kívül lesz hagyva, mert nem szerepel a célban. A egyeztetett sorok úgy frissülnek, hogy a last_updated mezőt az aktuális dátumra állítja. A séma a mező reviewhozzáadásához fejlődik. Az új sorokat az összes forrásmező felhasználásával szúrjuk be, kivéve a last_updated mezőt, amelyet null értékre állítunk.
Céloszlopok: id, title, last_updated
Forrásoszlopok: id, title, review, internal_count		 MERGE INTO target t
USING source s
ON t.id = s.id
WHEN MATCHED
THEN UPDATE SET last_updated = current_date()
WHEN NOT MATCHED
THEN INSERT * EXCEPT (last_updated, internal_count)		 INSERT hibát jelez, mert az oszlop internal_count nem létezik a céltáblában. A egyeztetett sorok úgy frissülnek, hogy a last_updated mezőt az aktuális dátumra állítja. A review mező hozzáadódik a céltáblához, de a internal_count mező figyelmen kívül lesz hagyva. Az újonnan beszúrt sorok last_updated értéke null van állítva.

Sémafejlődés engedélyezése Spark-konfigurációval (örökölt)

A Spark-konfigurációt spark.databricks.delta.schema.autoMerge.enabled úgy állíthatja be, hogy true engedélyezze a sémafejlődést az aktuális SparkSession összes írási műveletéhez:

Python

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", True)

Scala

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", true)

SQL

SET spark.databricks.delta.schema.autoMerge.enabled=true

Note

A Databricks nem javasolja ezt a megközelítést éles környezetben. A munkamenet-szintű konfiguráció beállítása több művelet nem szándékos sémamódosításaihoz vezethet, és megnehezíti annak okát, hogy mely műveletek fejlesztik a sémát.

Ehelyett engedélyezze a sémafejlődést minden egyes írási művelethez:

Ha beállításokat vagy szintaxist használ a sémafejlődés engedélyezéséhez egy írási műveletben, ez elsőbbséget élvez a Spark-konfigurációval szemben.

Táblaséma cseréje

Alapértelmezés szerint a tábla adatainak felülírása nem írja felül a sémát. Amikor a(z) mode("overwrite") használatával, replaceWhere nélkül ír felül egy táblát, előfordulhat, hogy az írandó adatok sémáját is felül szeretné írni.

A tábla sémájának és particionálásának lecseréléséhez állítsa a overwriteSchema beállítást true értékre:

df.write.option("overwriteSchema", "true")

Note

A dinamikus partíció felülírása esetén nem adható meg overwriteSchema mint true. Lásd a(z) dinamikus partíció felülírások (örökölt)partitionOverwriteMode elemet.

  • Last updated on