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_VERSIONdatabá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_VERSIONhodnoty 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_VERSIONhodnotami 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_VERSIONhodnotami) zsys.change_tracking_tablesdat ve zdrojové tabulce a pak je přesunete do cíle.Vytvořte jednu aktivitu uložené procedury , která
SYS_CHANGE_VERSIONaktualizuje 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_tablezdrojové ú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_RETENTIONvě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_versionvý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_VERSIONhodnoty 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_VERSIONhodnotami z Azure SQL Database do Azure Blob Storage. -
Aktivita uložené procedury aktualizuje hodnotu pro
SYS_CHANGE_VERSIONdalší 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_versionv 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_versiontabulce.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 CurrentTrackingVersionInt64 @{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: