Poznámka
Přístup k této stránce vyžaduje autorizaci. Můžete se zkusit přihlásit nebo změnit adresáře.
Přístup k této stránce vyžaduje autorizaci. Můžete zkusit změnit adresáře.
VZTAHUJE SE NA: Azure Data Factory
Azure Synapse Analytics
Návod
Vyzkoušejte si službu Data Factory v Microsoft Fabric, řešení pro analýzy typu all-in-one pro podniky. Microsoft Fabric zahrnuje vše od přesunu dat až po datovou vědu, analýzy v reálném čase, podnikovou inteligenci a reportování. Přečtěte si, jak začít používat novou zkušební verzi zdarma.
V řešení integrace dat je přírůstkové načítání dat po počátečním načtení dat široce používaným scénářem. Změněná data v určitém období ve zdrojovém úložišti dat lze snadno rozdělit na řezy (například LastModifyTime
). CreationTime
V některých případech však neexistuje žádný explicitní způsob, jak identifikovat rozdílová data z posledního zpracování dat. K identifikaci rozdílových dat můžete použít technologii sledování změn podporovanou úložišti dat, jako je Azure SQL Database a SQL Server.
Tento kurz popisuje, jak pomocí Azure Data Factory se sledováním změn přírůstkově načítat rozdílová data z Azure SQL Database do Azure Blob Storage. Další informace o sledování změn naleznete v tématu Sledování změn na serveru SQL Server.
V tomto kurzu provedete následující kroky:
- Připravte zdrojové úložiště dat.
- Vytvořte datovou továrnu.
- Vytvoření propojených služeb
- Vytvářejte zdrojové datové sady, datové sady jímky a sledovací datové sady.
- Vytvářejte, spouštějte a monitorujte celý kanál kopírování.
- Přidejte nebo aktualizujte data ve zdrojové tabulce.
- Vytvářejte, spouštějte a monitorujte kanál přírůstkového kopírování.
Řešení na vysoké úrovni
V tomto kurzu vytvoříte dva kanály, které provádějí následující operace.
Poznámka:
V tomto kurzu se jako zdrojové úložiště dat používá Azure SQL Database. Můžete také použít SQL Server.
Počáteční načtení historických dat: Vytvoříte kanál s aktivitou kopírování, která zkopíruje celá data ze zdrojového úložiště dat (Azure SQL Database) do cílového úložiště dat (Azure Blob Storage):
- Povolte technologii sledování změn ve zdrojové databázi v Azure SQL Database.
- Získejte počáteční hodnotu v
SYS_CHANGE_VERSION
databázi jako směrný plán pro zachycení změněných dat. - Načtěte úplná data ze zdrojové databáze do služby Azure Blob Storage.
Přírůstkové načítání rozdílových dat podle plánu: Vytvoříte kanál s následujícími aktivitami a pravidelně ho spouštíte:
Vytvořte dvě vyhledávací aktivity , které získají staré a nové
SYS_CHANGE_VERSION
hodnoty z Azure SQL Database.Vytvořte jednu aktivitu kopírování , která zkopíruje vložená, aktualizovaná nebo odstraněná data (rozdílová data) mezi těmito dvěma
SYS_CHANGE_VERSION
hodnotami z Azure SQL Database do Azure Blob Storage.Rozdílová data načtete spojením primárních klíčů změněných řádků (mezi dvěma
SYS_CHANGE_VERSION
hodnotami) zsys.change_tracking_tables
dat ve zdrojové tabulce a pak je přesunete do cíle.Vytvořte jednu aktivitu uložené procedury , která
SYS_CHANGE_VERSION
aktualizuje hodnotu pro další spuštění kanálu.
Požadavky
- Předplatné Azure. Pokud ho nemáte, vytvořte si bezplatný účet před tím, než začnete.
- Azure SQL Database. Jako zdrojové úložiště dat používáte databázi v Azure SQL Database. Pokud ho nemáte, přečtěte si článek Vytvoření databáze v Azure SQL Database , kde najdete postup jejího vytvoření.
- Účet úložiště Azure. Jako úložiště dat jímky používáte službu Blob Storage. Pokud účet úložiště Azure nemáte, přečtěte si článek Vytvoření účtu úložiště , kde najdete postup jeho vytvoření. Vytvořte kontejner s názvem adftutorial.
Poznámka:
K interakci s Azure doporučujeme použít modul Azure Az PowerShell. Pokud chcete začít, přečtěte si téma Instalace Azure PowerShellu. Informace o tom, jak migrovat na modul Az PowerShell, najdete v tématu Migrace Azure PowerShellu z AzureRM na Az.
Vytvoření tabulky zdroje dat v Azure SQL Database
Otevřete SQL Server Management Studio a připojte se k SQL Database.
V Průzkumník serveru klikněte pravým tlačítkem myši na databázi a pak vyberte Nový dotaz.
Spuštěním následujícího příkazu SQL pro vaši databázi vytvořte tabulku s názvem
data_source_table
zdrojové úložiště dat:create table data_source_table ( PersonID int NOT NULL, Name varchar(255), Age int PRIMARY KEY (PersonID) ); INSERT INTO data_source_table (PersonID, Name, Age) VALUES (1, 'aaaa', 21), (2, 'bbbb', 24), (3, 'cccc', 20), (4, 'dddd', 26), (5, 'eeee', 22);
Povolte sledování změn v databázi a zdrojové tabulce (
data_source_table
) spuštěním následujícího dotazu SQL.Poznámka:
- Nahraďte
<your database name>
názvem databáze v Azure SQL Database, která mádata_source_table
. - Změněná data jsou v tomto příkladu uchovávána po dobu dvou dnů. Pokud načtete změněná data každé tři dny nebo častěji, některá změněná data nebudou zahrnuta. Musíte buď změnit hodnotu na
CHANGE_RETENTION
větší číslo, nebo zajistit, aby vaše lhůta pro načtení změněných dat byla do dvou dnů. Další informace naleznete v tématu Povolení sledování změn pro databázi.
ALTER DATABASE <your database name> SET CHANGE_TRACKING = ON (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON) ALTER TABLE data_source_table ENABLE CHANGE_TRACKING WITH (TRACK_COLUMNS_UPDATED = ON)
- Nahraďte
Vytvořte novou tabulku a uložte ji s
ChangeTracking_version
výchozí hodnotou spuštěním následujícího dotazu:create table table_store_ChangeTracking_version ( TableName varchar(255), SYS_CHANGE_VERSION BIGINT, ); DECLARE @ChangeTracking_version BIGINT SET @ChangeTracking_version = CHANGE_TRACKING_CURRENT_VERSION(); INSERT INTO table_store_ChangeTracking_version VALUES ('data_source_table', @ChangeTracking_version)
Poznámka:
Pokud se data po povolení sledování změn pro SQL Database nezmění, hodnota verze sledování změn je
0
.Spuštěním následujícího dotazu vytvořte uloženou proceduru ve své databázi. Kanál vyvolá tuto uloženou proceduru, aby aktualizoval verzi sledování změn v tabulce, kterou jste vytvořili v předchozím kroku.
CREATE PROCEDURE Update_ChangeTracking_Version @CurrentTrackingVersion BIGINT, @TableName varchar(50) AS BEGIN UPDATE table_store_ChangeTracking_version SET [SYS_CHANGE_VERSION] = @CurrentTrackingVersion WHERE [TableName] = @TableName END
Vytvoření datové továrny
Otevřete webový prohlížeč Microsoft Edge nebo Google Chrome. V současné době podporují uživatelské rozhraní služby Data Factory pouze tyto prohlížeče.
V Azure Portal v nabídce vlevo vyberte Vytvořit prostředek.
Vyberte možnost Analytics>Data Factory :
Na stránce Nová datová továrna zadejte jako název ADFTutorialDataFactory .
Název datové továrny musí být globálně jedinečný. Pokud se zobrazí chybová zpráva, že název, který jste vybrali, není k dispozici, změňte název (například na yournameADFTutorialDataFactory) a zkuste datovou továrnu vytvořit znovu. Další informace najdete v tématu Pravidla pojmenování Azure Data Factory.
Vyberte předplatné Azure, ve kterém chcete datovou továrnu vytvořit.
U položky Skupina prostředků proveďte jeden z následujících kroků:
- Vyberte možnost Použít existující a poté z rozevíracího seznamu vyberte existující skupinu prostředků.
- Vyberte Vytvořit novou a poté zadejte název skupiny prostředků.
Informace o skupinách prostředků najdete v článku Použití skupin prostředků ke správě prostředků Azure.
Jako Verzi vyberte V2.
V části Oblast vyberte oblast pro datovou továrnu.
V rozevíracím seznamu se zobrazují pouze podporovaná umístění. Úložiště dat (například Azure Storage a Azure SQL Database) a výpočetní prostředky (například Azure HDInsight), které datová továrna používá, mohou být v jiných oblastech.
Vyberte Další: Konfigurace Gitu. Nastavte úložiště podle pokynů v části Metoda konfigurace 4: Během vytváření továrny nebo zaškrtněte políčko Konfigurovat Git později .
Vyberte možnost Zkontrolovat a vytvořit.
Vyberte Vytvořit.
Na řídicím panelu je na dlaždici Nasazení služby Data Factory zobrazen stav.
Po dokončení vytváření se zobrazí stránka Data Factory . Výběrem dlaždice Spustit studio otevřete uživatelské rozhraní Azure Data Factory na samostatné kartě.
Vytvoření propojených služeb
V datové továrně vytvoříte propojené služby, abyste svá úložiště dat a výpočetní služby spojili s datovou továrnou. V této části vytvoříte propojené služby k vašemu účtu úložiště Azure a databázi ve službě Azure SQL Database.
Vytvoření propojené služby Azure Storage
Propojení účtu úložiště s datovou továrnou:
- V uživatelském rozhraní služby Data Factory na kartě Spravovat v části Připojení vyberte Propojené služby. Poté vyberte + Nový nebo tlačítko Vytvořit propojenou službu .
- V okně Nová propojená služba vyberte Azure Blob Storage a pak vyberte Pokračovat.
- Zadejte následující informace:
- Jako Název zadejte AzureStorageLinkedService.
- V části Připojení prostřednictvím prostředí Integration Runtime vyberte prostředí Integration Runtime.
- V poli Typ ověření vyberte metodu ověřování.
- Jako Název účtu úložiště vyberte svůj účet úložiště Azure.
- Vyberte Vytvořit.
Vytvoření propojené služby Azure SQL Database
Propojení databáze s datovou továrnou:
V uživatelském rozhraní služby Data Factory na kartě Spravovat v části Připojení vyberte Propojené služby. Poté vyberte + Nový.
V okně Nová propojená služba vyberte Azure SQL Database a pak vyberte Pokračovat.
Zadejte následující informace:
- Do pole Název zadejte AzureSqlDatabaseLinkedService.
- V poli Název serveru vyberte svůj server.
- V poli Název databáze vyberte svou databázi.
- V poli Typ ověření vyberte metodu ověřování. V tomto kurzu se k demonstraci používá ověřování SQL.
- Do pole Uživatelské jméno zadejte jméno uživatele.
- Do pole Heslo zadejte heslo pro uživatele. Nebo zadejte informace o propojené službě Azure Key Vault – AKV, názvu tajného klíče a verzi tajného klíče.
Výběrem možnosti Test připojení otestujte připojení.
Vyberte Vytvořit a vytvořte propojenou službu.
Vytvoření datových sad
V této části vytvoříte datové sady, které budou představovat zdroj dat a cíl dat spolu s místem pro uložení SYS_CHANGE_VERSION
hodnot.
Vytvoření datové sady pro reprezentaci zdrojových dat
V uživatelském rozhraní služby Data Factory na kartě Autor vyberte znaménko plus (+). Pak vyberte Datová sada nebo vyberte tři tečky pro akce datové sady.
Vyberte Azure SQL Database a pak vyberte Pokračovat.
V okně Nastavit vlastnosti proveďte následující kroky:
- Do pole Název zadejte SourceDataset.
- V části Propojená služba vyberte AzureSqlDatabaseLinkedService.
- Jako Název tabulky vyberte dbo.data_source_table.
- V části Importovat schéma vyberte možnost Z připojení/úložiště .
- Vyberte OK.
Vytvoření datové sady pro reprezentaci dat zkopírovaných do úložiště dat jímky
V následujícím postupu vytvoříte datovou sadu, která bude představovat data zkopírovaná ze zdrojového úložiště dat. Kontejner adftutorial jste vytvořili v Azure Blob Storage jako součást předpokladů. Vytvořte kontejner, pokud neexistuje, nebo jej nastavte na název existujícího kontejneru. V tomto kurzu je název výstupního souboru dynamicky generován z výrazu @CONCAT('Incremental-', pipeline().RunId, '.txt')
.
V uživatelském rozhraní služby Data Factory na kartě Vytvořit vyberte +možnost . Pak vyberte Datová sada nebo vyberte tři tečky pro akce datové sady.
Vyberte Azure Blob Storage a pak vyberte Pokračovat.
Vyberte formát datového typu jako DelimitedText a poté vyberte Pokračovat.
V okně Nastavit vlastnosti proveďte následující kroky:
- Jako Název zadejte SinkDataset.
- V části Propojená služba vyberte AzureBlobStorageLinkedService.
- Jako Cesta k souboru zadejte adftutorial/incchgtracking.
- Vyberte OK.
Jakmile se datová sada zobrazí ve stromovém zobrazení, přejděte na kartu Připojení a vyberte textové pole Název souboru . Když se zobrazí možnost Přidat dynamický obsah , vyberte ji.
Zobrazí se okno Tvůrce výrazů kanálu . Vložte
@concat('Incremental-',pipeline().RunId,'.csv')
do textového pole.Vyberte OK.
Vytvoření datové sady pro reprezentaci dat sledování změn
V následujícím postupu vytvoříte datovou sadu pro uložení verze sledování změn. Tabulku table_store_ChangeTracking_version
jste vytvořili jako součást předpokladů.
- V uživatelském rozhraní služby Data Factory na kartě Autor vyberte +a pak vyberte Datová sada.
- Vyberte Azure SQL Database a pak vyberte Pokračovat.
- V okně Nastavit vlastnosti proveďte následující kroky:
- Do pole Název zadejte ChangeTrackingDataset.
- V části Propojená služba vyberte AzureSqlDatabaseLinkedService.
- V poli Název tabulky vyberte možnost dbo.table_store_ChangeTracking_version.
- V části Importovat schéma vyberte možnost Z připojení/úložiště .
- Vyberte OK.
Vytvoření kanálu pro úplnou kopii
V následujícím postupu vytvoříte kanál s aktivitou kopírování, která zkopíruje celá data ze zdrojového úložiště dat (Azure SQL Database) do cílového úložiště dat (Azure Blob Storage):
V uživatelském rozhraní služby Data Factory na kartě Vytvořit vyberte +možnost a poté vyberte možnost Kanál kanálu>.
Zobrazí se nová karta pro konfiguraci kanálu. Kanál se zobrazí také ve stromovém zobrazení. V okně Vlastnosti změňte název kanálu na FullCopyPipeline.
Na panelu nástrojů Činnosti rozbalte položku Přesunout a transformovat. Proveďte jeden z následujících kroků:
- Přetáhněte aktivitu kopírování na plochu návrháře kanálu.
- Na panelu hledání v části Aktivity vyhledejte aktivitu kopírování dat a poté nastavte název na FullCopyActivity.
Přepněte na kartu Zdroj . V části Zdrojová datová sada vyberte SourceDataset.
Přepněte na kartu Jímka . V části Datová sada jímky vyberte SinkDataset.
Chcete-li ověřit definici kanálu, vyberte na panelu nástrojů možnost Ověřit . Ověřte, že nedošlo k chybě ověření. Zavřete výstup ověření kanálu.
Chcete-li publikovat entity (propojené služby, datové sady a kanály), vyberte možnost Publikovat vše. Počkejte, dokud se nezobrazí zpráva Publikování proběhlo úspěšně.
Chcete-li zobrazit oznámení, vyberte tlačítko Zobrazit oznámení .
Spuštění úplného kanálu kopií
V uživatelském rozhraní služby Data Factory na panelu nástrojů kanálu vyberte Přidat spouštěč a pak vyberte Spustit nyní.
V okně Spuštění kanálu vyberte OK.
Monitorování úplného kanálu kopírování
V uživatelském rozhraní služby Data Factory vyberte kartu Monitorování . Spuštění kanálu a jeho stav se zobrazí v seznamu. Chcete-li seznam aktualizovat, vyberte možnost Obnovit. Najeďte myší na spuštění kanálu a získejte možnost Znovu spustit nebo Consumption .
Pokud chcete zobrazit spuštění aktivit přidružená ke spuštění kanálu, vyberte název kanálu ve sloupci Název kanálu . V kanálu je pouze jedna aktivita, takže v seznamu je pouze jedna položka. Pokud chcete přepnout zpět na zobrazení spuštění kanálu, vyberte odkaz Všechna spuštění kanálu v horní části.
Kontrola výsledků
Složka incchgtracking kontejneru adftutorial obsahuje soubor s názvem incremental-<GUID>.csv
.
Soubor by měl obsahovat data z vaší databáze:
PersonID,Name,Age
1,"aaaa",21
2,"bbbb",24
3,"cccc",20
4,"dddd",26
5,"eeee",22
5,eeee,PersonID,Name,Age
1,"aaaa",21
2,"bbbb",24
3,"cccc",20
4,"dddd",26
5,"eeee",22
Přidání dalších dat do zdrojové tabulky
Spuštěním následujícího dotazu na databázi přidejte řádek a aktualizujte řádek:
INSERT INTO data_source_table
(PersonID, Name, Age)
VALUES
(6, 'new','50');
UPDATE data_source_table
SET [Age] = '10', [name]='update' where [PersonID] = 1
Vytvoření kanálu pro rozdílovou kopii
V následujícím postupu vytvoříte kanál s aktivitami a pravidelně ho spouštíte. Když spustíte kanál:
-
Aktivity vyhledávání získávají staré a nové
SYS_CHANGE_VERSION
hodnoty z Azure SQL Database a předávají je aktivitě kopírování. -
Aktivita kopírování zkopíruje vložená, aktualizovaná nebo odstraněná data mezi těmito dvěma
SYS_CHANGE_VERSION
hodnotami z Azure SQL Database do Azure Blob Storage. -
Aktivita uložené procedury aktualizuje hodnotu pro
SYS_CHANGE_VERSION
další spuštění kanálu.
V uživatelském rozhraní služby Data Factory přepněte na kartu Autor. Vyberte možnost +a poté vyberte možnost Kanál kanálu>.
Zobrazí se nová karta pro konfiguraci kanálu. Kanál se zobrazí také ve stromovém zobrazení. V okně Vlastnosti změňte název kanálu na IncrementalCopyPipeline.
Rozbalte položku Obecné v panelu nástrojů Činnosti . Přetáhněte aktivitu vyhledávání na plochu návrháře kanálu nebo vyhledejte v poli Aktivity vyhledávání . Nastavte název aktivity na LookupLastChangeTrackingVersionActivity. Tato aktivita získá verzi sledování změn použitou v poslední operaci kopírování, která je uložená
table_store_ChangeTracking_version
v tabulce.Přepněte na kartu Nastavení v okně Vlastnosti . V části Zdrojová datová sada vyberte ChangeTrackingDataset.
Přetáhněte vyhledávací aktivitu z panelu nástrojů Aktivity na plochu návrháře kanálu. Nastavte název aktivity na LookupCurrentChangeTrackingVersionActivity. Tato aktivita získá aktuální verzi sledování změn.
Přepněte na kartu Nastavení v okně Vlastnosti a proveďte následující kroky:
V části Zdrojová datová sada vyberte SourceDataset.
V části Použít dotaz vyberte možnost Dotaz.
Do pole Dotaz zadejte následující dotaz SQL:
SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion
Na panelu nástrojů Činnosti rozbalte položku Přesunout a transformovat. Přetáhněte aktivitu kopírování dat na plochu návrháře kanálu. Nastavte název aktivity na IncrementalCopyActivity. Tato aktivita zkopíruje data mezi poslední verzí sledování změn a aktuální verzí sledování změn do cílového úložiště dat.
Přepněte na kartu Zdroj v okně Vlastnosti a proveďte následující kroky:
V části Zdrojová datová sada vyberte SourceDataset.
V části Použít dotaz vyberte možnost Dotaz.
Do pole Dotaz zadejte následující dotaz SQL:
SELECT data_source_table.PersonID,data_source_table.Name,data_source_table.Age, CT.SYS_CHANGE_VERSION, SYS_CHANGE_OPERATION from data_source_table RIGHT OUTER JOIN CHANGETABLE(CHANGES data_source_table, @{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.SYS_CHANGE_VERSION}) AS CT ON data_source_table.PersonID = CT.PersonID where CT.SYS_CHANGE_VERSION <= @{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
Přepněte na kartu Jímka . V části Datová sada jímky vyberte SinkDataset.
Připojte obě vyhledávací aktivity k aktivitě kopírování jednu po druhé. Přetáhněte zelené tlačítko, které je připojeno k aktivitě vyhledávání, do aktivity kopírování.
Přetáhněte uloženou aktivitu procedury z panelu nástrojů Aktivity na plochu návrháře kanálů. Nastavte název aktivity na StoredProceduretoUpdateChangeTrackingActivity. Tato aktivita aktualizuje verzi sledování změn v
table_store_ChangeTracking_version
tabulce.Přepněte na kartu Nastavení a proveďte následující kroky:
- V části Propojená služba vyberte AzureSqlDatabaseLinkedService.
- V části Název uložené procedury vyberte možnost Update_ChangeTracking_Version.
- Vyberte Importovat.
- V části Parametry uložené procedury zadejte následující hodnoty parametrů:
Název Typ Hodnota CurrentTrackingVersion
Int64 @{activity('LookupCurrentChangeTrackingVersionActivity').output.firstRow.CurrentChangeTrackingVersion}
TableName
Řetězec @{activity('LookupLastChangeTrackingVersionActivity').output.firstRow.TableName}
Připojte aktivitu kopírování k aktivitě uložené procedury. Přetáhněte zelené tlačítko, které je připojeno k aktivitě kopírování, do aktivity uložené procedury.
Na panelu nástrojů vyberte možnost Ověřit . Ověřte, že nejsou žádné chyby při validaci. Zavřete okno Sestava ověření kanálu .
Publikujte entity (propojené služby, datové sady a kanály) do služby Data Factory výběrem tlačítka Publikovat vše . Počkejte, dokud se nezobrazí zpráva Publikování proběhlo úspěšně .
Spuštění kanálu přírůstkového kopírování
Na panelu nástrojů pro kanál vyberte Přidat trigger a poté vyberte Aktivovat nyní.
V okně Spuštění kanálu vyberte OK.
Monitorování kanálu přírůstkového kopírování
Vyberte kartu Monitor . Spuštění kanálu a jeho stav se zobrazí v seznamu. Chcete-li seznam aktualizovat, vyberte možnost Obnovit.
Pokud chcete zobrazit spuštění aktivit přidružená ke spuštění kanálu, vyberte odkaz IncrementalCopyPipeline ve sloupci Název kanálu . Spuštění aktivit se zobrazí v seznamu.
Kontrola výsledků
Druhý soubor se zobrazí ve složce incchgtracking kontejneru adftutorial .
Soubor by měl obsahovat pouze rozdílová data z vaší databáze. Záznam s U
je aktualizovaný řádek v databázi a I
je to jeden přidaný řádek.
PersonID,Name,Age,SYS_CHANGE_VERSION,SYS_CHANGE_OPERATION
1,update,10,2,U
6,new,50,1,I
První tři sloupce jsou změněná data z data_source_table
. Poslední dva sloupce jsou metadata z tabulky pro systém sledování změn. Čtvrtý sloupec je SYS_CHANGE_VERSION
hodnota pro každý změněný řádek. V pátém sloupci je operace: U
= update, I
= insert. Podrobnosti o informacích o sledování změn naleznete v části CHANGETABLE.
==================================================================
PersonID Name Age SYS_CHANGE_VERSION SYS_CHANGE_OPERATION
==================================================================
1 update 10 2 U
6 new 50 1 I
Související obsah
V následujícím výukovém programu se dozvíte, jak kopírovat pouze nové a změněné soubory na základě LastModifiedDate
: