table_changes
tabelwaardefunctie
Van toepassing op: Databricks Databricks Runtime
Retourneert een logboek met wijzigingen in een Delta Lake-tabel waarvoor wijzigingsgegevensfeed is ingeschakeld.
Als u deze functie wilt aanroepen, moet u beschikken over ten minste een van de volgende opties:
SELECT
bevoegdheden voor de opgegeven tabel- De eigenaar van de tabel zijn
- Beheerdersbevoegdheden hebben
Syntaxis
table_changes ( table_str, start [, end ] )
Argumenten
table_str
: een letterlijke tekenreeks die de optioneel gekwalificeerde naam van de tabel aangeeft.start
: een letterlijke BIGINT of TIMESTAMP, die de eerste versie of tijdstempel van de wijziging vertegenwoordigt die moet worden geretourneerd.end
: Een optionele letterlijke BIGINT of TIMESTAMP, die de laatste versie of tijdstempel van de wijziging aangeeft die moet worden geretourneerd. Als dit niet is opgegeven, worden alle wijzigingen vanstart
tot en met de huidige wijziging geretourneerd.
Retourneert
Een tabel met alle kolommen van de tabel die zijn geïdentificeerd in table_str
, plus de volgende kolommen:
_change_type STRING NOT NULL
Hiermee geeft u de wijziging op:
delete
,insert
,update_preimage
ofupdate_postimage
_commit_version BIGINT NOT NULL
Hiermee geeft u de doorvoerversie van de tabel die is gekoppeld aan de wijziging.
_commit_timestamp TIMESTAMP NOT NULL
Hiermee geeft u de doorvoertijdstempel op die is gekoppeld aan de wijziging.
Als table_str
geen gekwalificeerde tabelnaam vertegenwoordigt, wordt de naam gekwalificeerd met de waarde van current_schema
.
Als de tabelnaam spaties of puntjes bevat, gebruikt u aanhalingstekens in de tekenreeks om dat deel van de naam aan te citeren.
Voorbeelden
-- Create a Delta table with Change Data Feed;
> CREATE TABLE myschema.t(c1 INT, c2 STRING) TBLPROPERTIES(delta.enableChangeDataFeed=true);
-- Modify the table
> INSERT INTO myschema.t VALUES (1, 'Hello'), (2, 'World');
> INSERT INTO myschema.t VALUES (3, '!');
> UPDATE myschema.t SET c2 = upper(c2) WHERE c1 < 3;
> DELETE FROM myschema.t WHERE c1 = 3;
-- Show the history of table change events
> DESCRIBE HISTORY myschema.t;
version timestamp userId userName operation operationParameters ...
4 2022-09-01T18:32:35.000+0000 6167625779053302 alf@melmak.et DELETE {"predicate":"[\"(spark_catalog.myschema.t.c1 = 3)\"]"}
3 2022-09-01T18:32:32.000+0000 6167625779053302 alf@melmak.et UPDATE {"predicate":"(c1#3195878 < 3)"}
2 2022-09-01T18:32:28.000+0000 6167625779053302 alf@melmak.et WRITE {"mode":"Append","partitionBy":"[]"}
1 2022-09-01T18:32:26.000+0000 6167625779053302 alf@melmak.et WRITE {"mode":"Append","partitionBy":"[]"}
0 2022-09-01T18:32:23.000+0000 6167625779053302 alf@melmak.et CREATE TABLE {"isManaged":"true","description":null,"partitionBy":"[]","properties":"{\"delta.enableChangeDataFeed\":\"true\"}"}
-- Show the change table feed using a the commit timestamp retrieved from the history.
> SELECT * FROM table_changes('`myschema`.`t`', 2);
c1 c2 _change_type _commit_version _commit_timestamp
3 ! insert 2 2022-09-01T18:32:28.000+0000
2 WORLD update_postimage 3 2022-09-01T18:32:32.000+0000
2 World update_preimage 3 2022-09-01T18:32:32.000+0000
1 Hello update_preimage 3 2022-09-01T18:32:32.000+0000
1 HELLO update_postimage 3 2022-09-01T18:32:32.000+0000
3 ! delete 4 2022-09-01T18:32:35.000+0000
-- Show the ame change table feed using a point in time.
> SELECT * FROM table_changes('`myschema`.`t`', '2022-09-01T18:32:27.000+0000') ORDER BY _commit_version;
c1 c2 _change_type _commit_version _commit_timestamp
3 ! insert 2 2022-09-01T18:32:28.000+0000
2 WORLD update_postimage 3 2022-09-01T18:32:32.000+0000
2 World update_preimage 3 2022-09-01T18:32:32.000+0000
1 Hello update_preimage 3 2022-09-01T18:32:32.000+0000
1 HELLO update_postimage 3 2022-09-01T18:32:32.000+0000
3 ! delete 4 2022-09-01T18:32:35.000+0000