Adatok növekményes betöltése az SQL Server több táblájából egy Azure SQL Database-adatbázisba az Azure Portal használatával

  • Cikk

A következőkre vonatkozik: Azure Data Factory Azure Synapse Analytics

Tipp.

Próbálja ki a Data Factoryt a Microsoft Fabricben, amely egy teljes körű elemzési megoldás a nagyvállalatok számára. A Microsoft Fabric az adattovábbítástól az adatelemzésig, a valós idejű elemzésig, az üzleti intelligenciáig és a jelentéskészítésig mindent lefed. Ismerje meg, hogyan indíthat új próbaverziót ingyenesen!

Ebben az oktatóanyagban létrehoz egy Azure Data Factoryt egy folyamattal, amely egy SQL Server-adatbázis több táblájából betölti a deltaadatokat egy Azure SQL Database-adatbázisba.

Az oktatóanyagban az alábbi lépéseket fogja végrehajtani:

  • Forrás- és céladattárak előkészítése.
  • Adat-előállító létrehozása
  • Create a self-hosted integration runtime.
  • Az Integration Runtime telepítése.
  • Társított szolgáltatások létrehozása.
  • Forrás-, fogadó- és küszöbadatkészletek létrehozása.
  • Folyamat létrehozása, futtatása és figyelése.
  • Tekintse át az eredményeket.
  • Adatok hozzáadása vagy frissítése a forrástáblákban.
  • A folyamat újrafuttatása és monitorozása.
  • A végső eredmények áttekintése.

Áttekintés

Az alábbiak a megoldás kialakításának leglényegesebb lépései:

  1. A küszöb oszlopának kiválasztása.

    Jelölje ki az adatforrás egyes táblázatainak egy-egy oszlopát, amely alapján az új és a frissített rekordok minden egyes futtatáskor azonosíthatóak. Normális esetben az ebben a kiválasztott oszlopban (például: last_modify_time vagy ID) lévő adatok a sorok létrehozásával vagy frissítésével folyamatosan növekednek. Az ebben az oszlopban lévő legnagyobb érték szolgál a küszöbként.

  2. Egy adatraktár előkészítése a küszöbértékek tárolására.

    Ebben az oktatóanyagban a küszöbértékeket egy SQL-adatbázisban tároljuk.

  3. Egy folyamat létrehozása a következő tevékenységekkel:

    a. Egy ForEach tevékenység létrehozása, amely végighalad a forrástáblanevek listáján, amelyet a rendszer paraméterként ad át a folyamatnak. Minden forrástáblához elindítja a következő tevékenységeket a változásadatok betöltéséhez az adott tábla esetében.

    b. Két keresési tevékenység létrehozása. Az első keresési tevékenység az utolsó küszöbértéket kéri le. A második keresési tevékenység az új küszöbértéket kéri le. Ezeket a küszöbértékeket a rendszer átadja a másolási tevékenységnek.

    c. Egy másolási tevékenység létrehozása, amely a küszöbértéket tartalmazó oszlopban a régi küszöbértéknél magasabb, az új küszöbértéknél alacsonyabb értékkel rendelkező sorokat másolja át a forrásadattárból. Ezután a módosított adatokat a forrásadattárból új fájlként az Azure Blob Storage-ba másolja.

    d. Egy StoredProcedure tevékenység létrehozása, amely frissíti a küszöbértékeket a folyamat következő futtatásához.

    Itt látható a megoldás összefoglaló jellegű ábrája:

    Incrementally load data

Ha még nincs Azure-előfizetése, kezdés előtt hozzon létre egy ingyenes fiókot.

Előfeltételek

  • Egy SQL Server. Ebben az oktatóanyagban egy SQL Server-adatbázist használ forrásadattárként.
  • Azure SQL Database Az Azure SQL Database-ben adatbázist használ fogadóadattárként. Ha nem rendelkezik adatbázissal az SQL Database-ben, a létrehozás lépéseit az Adatbázis létrehozása az Azure SQL Database-ben című témakörben találja.

Forrástáblák létrehozása az SQL Server-adatbázisban

  1. Nyissa meg az SQL Server Management Studiót, és csatlakozzon az SQL Server-adatbázishoz.

  2. A Kiszolgálókezelőben kattintson a jobb gombbal az adatbázisra, és válassza az Új lekérdezés elemet.

  3. Futtassa a következő SQL-parancsot az adatbázison a customer_table és a project_table nevű tábla létrehozásához:

    create table customer_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

create table project_table
(
    Project varchar(255),
    Creationtime datetime
);

INSERT INTO customer_table
(PersonID, Name, LastModifytime)
VALUES
(1, 'John','9/1/2017 12:56:00 AM'),
(2, 'Mike','9/2/2017 5:23:00 AM'),
(3, 'Alice','9/3/2017 2:36:00 AM'),
(4, 'Andy','9/4/2017 3:21:00 AM'),
(5, 'Anny','9/5/2017 8:06:00 AM');

INSERT INTO project_table
(Project, Creationtime)
VALUES
('project1','1/1/2015 0:00:00 AM'),
('project2','2/2/2016 1:23:00 AM'),
('project3','3/4/2017 5:16:00 AM');

Céltáblák létrehozása az adatbázisban

  1. Nyissa meg az SQL Server Management Studiót, és csatlakozzon az adatbázishoz az Azure SQL Database-ben.

  2. A Kiszolgálókezelőben kattintson a jobb gombbal az adatbázisra, és válassza az Új lekérdezés elemet.

  3. Futtassa a következő SQL-parancsot az adatbázison a customer_table és a project_table nevű tábla létrehozásához:

    create table customer_table
(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

create table project_table
(
    Project varchar(255),
    Creationtime datetime
);

Hozzon létre egy másik táblát az adatbázisban a magas vízjel értékének tárolásához

  1. Futtassa a következő SQL-parancsot az adatbázison a vízjel értékének tárolására elnevezett watermarktable tábla létrehozásához:

    create table watermarktable
(

    TableName varchar(255),
    WatermarkValue datetime,
);

  2. Szúrja be a két forrástábla kezdeti küszöbértékeit a küszöbértékek táblájába.

    
INSERT INTO watermarktable
VALUES 
('customer_table','1/1/2010 12:00:00 AM'),
('project_table','1/1/2010 12:00:00 AM');

Tárolt eljárás létrehozása az adatbázisban

Futtassa a következő parancsot egy tárolt eljárás létrehozásához az adatbázisban. Ez a tárolt eljárás minden folyamatfuttatás után frissíti a küszöbértéket.

CREATE PROCEDURE usp_write_watermark @LastModifiedtime datetime, @TableName varchar(50)
AS

BEGIN

UPDATE watermarktable
SET [WatermarkValue] = @LastModifiedtime 
WHERE [TableName] = @TableName

END

Adattípusok és további tárolt eljárások létrehozása az adatbázisban

Futtassa a következő lekérdezést két tárolt eljárás és két adattípus létrehozásához az adatbázisban. Ezek összevonják a forrástáblák adatait a céltáblákba.

Annak érdekében, hogy az utazás könnyen elindulhasson, közvetlenül ezeket a tárolt eljárásokat használjuk, amelyek egy táblaváltozón keresztül adják át a deltaadatokat, majd egyesítjük őket a céltárolóba. Ne feledje, hogy nem számít arra, hogy a táblaváltozóban "nagy" számú (100-nál több) deltasort kell tárolni.

Ha nagy számú deltasort kell egyesítenie a céltárolóba, javasoljuk, hogy másolási tevékenység használatával másolja az összes delta-adatot egy ideiglenes "átmeneti" táblába a céltárolóban, majd a táblaváltozó használata nélkül saját tárolt eljárást épített ki az "előkészítési" táblából a "végleges" táblába való egyesítéshez.

CREATE TYPE DataTypeforCustomerTable AS TABLE(
    PersonID int,
    Name varchar(255),
    LastModifytime datetime
);

GO

CREATE PROCEDURE usp_upsert_customer_table @customer_table DataTypeforCustomerTable READONLY
AS

BEGIN
  MERGE customer_table AS target
  USING @customer_table AS source
  ON (target.PersonID = source.PersonID)
  WHEN MATCHED THEN
      UPDATE SET Name = source.Name,LastModifytime = source.LastModifytime
  WHEN NOT MATCHED THEN
      INSERT (PersonID, Name, LastModifytime)
      VALUES (source.PersonID, source.Name, source.LastModifytime);
END

GO

CREATE TYPE DataTypeforProjectTable AS TABLE(
    Project varchar(255),
    Creationtime datetime
);

GO

CREATE PROCEDURE usp_upsert_project_table @project_table DataTypeforProjectTable READONLY
AS

BEGIN
  MERGE project_table AS target
  USING @project_table AS source
  ON (target.Project = source.Project)
  WHEN MATCHED THEN
      UPDATE SET Creationtime = source.Creationtime
  WHEN NOT MATCHED THEN
      INSERT (Project, Creationtime)
      VALUES (source.Project, source.Creationtime);
END

Adat-előállító létrehozása

  1. Indítsa el a Microsoft Edge vagy a Google Chrome böngészőt. A Data Factory felhasználói felületének használata jelenleg csak a Microsoft Edge-ben és a Google Chrome-ban támogatott.

  2. A bal oldali menüben válassza az Erőforrás-integrációs>>adat-előállító létrehozása lehetőséget:

    Data Factory selection in the "New" pane

  3. Az Új data factory lapon, a Név mezőben adja meg a következőt: ADFMultiIncCopyTutorialDF.

    Az Azure Data Factory nevének globálisan egyedinek kell lennie. Ha egy piros felkiáltójelet lát a következő hibaüzenettel, változtassa meg az adat-előállító nevét (például a következőre: sajátneveADFIncCopyTutorialDF), majd próbálkozzon újra a létrehozással. A Data Factory-összetevők részleteit a Data Factory elnevezési szabályait ismertető cikkben találja.

    Data factory name "ADFIncCopyTutorialDF" is not available

  4. Válassza ki azt az Azure-előfizetést, amelyben az adat-előállítót létre szeretné hozni.

  5. Az erőforráscsoportban hajtsa végre a következő lépések egyikét:

    • Kattintson a Meglévő használata elemre, majd a legördülő listából válasszon egy meglévő erőforráscsoportot.
    • Kattintson az Új létrehozása elemre, és adja meg az erőforráscsoport nevét.
      Az erőforráscsoportokkal kapcsolatos információkért tekintse meg a Using resource groups to manage your Azure resources (Erőforráscsoportok használata az Azure-erőforrások kezeléséhez) című cikket.

  6. A Verzió résznél válassza a V2 értéket.

  7. Válassza ki a Data Factory helyét. A legördülő listán csak a támogatott helyek jelennek meg. Az adat-előállítók által használt adattárak (Azure Storage, Azure SQL Database stb.) és számítási erőforrások (HDInsight stb.) más régiókban is lehetnek.

  8. Kattintson a Létrehozás gombra.

  9. A létrehozás befejezése után a Data Factory lap a képen látható módon jelenik meg.

    Home page for the Azure Data Factory, with the Open Azure Data Factory Studio tile.

  10. Válassza a Megnyitás az Azure Data Factory Studio megnyitása csempén az Azure Data Factory felhasználói felületének (UI) külön lapon való elindításához.

Saját üzemeltetésű integrációs modul létrehozása

Mialatt adatokat helyez át egy magánhálózaton (helyszínen) lévő adattárból egy Azure-adattárba, telepítsen egy saját üzemeltetésű integrációs modult (IR) a helyszíni környezetben. A saját üzemeltetésű integrációs modul adatokat helyez át a magánhálózat és az Azure között.

  1. Az Azure Data Factory felhasználói felületének kezdőlapján válassza a Bal szélső panel Kezelés lapját .

    The home page Manage button

  2. A bal oldali panelen válassza az Integrációs modulokat , majd az +Új lehetőséget.

    Create an integration runtime

  3. Az Integrációs modul beállítása ablakban válassza az Adatáthelyezési és -küldési tevékenységek végrehajtása külső számításoknak lehetőséget, majd kattintson a Folytatás gombra.

  4. Válassza a Saját üzemeltetésű lehetőséget, majd kattintson a Folytatás gombra.

  5. Írja be a MySelfHostedIR nevet, és kattintson a Létrehozás gombra.

  6. Kattintson a Kattintson ide a számítógépen történő expressz telepítés indításához elemre az 1. lehetőség: Expressz telepítés szakaszban.

    Click Express setup link

  7. Az Integrációs modul (Saját üzemeltetésű) – Expressz telepítés ablakban kattintson a Bezárás elemre.

    Integration runtime setup - successful

  8. A webböngészőben kattintson a Befejezés gombra az Integrációs modul telepítése ablakban.

  9. Győződjön meg róla, hogy a MySelfHostedIR szerepel az integrációs modulok listájában.

Társított szolgáltatások létrehozása

Társított szolgáltatásokat hoz létre egy adat-előállítóban az adattárak és a számítási szolgáltatások adat-előállítóval történő társításához. Ebben a szakaszban társított szolgáltatásokat hoz létre az SQL Server-adatbázishoz és az Adatbázishoz az Azure SQL Database-ben.

Az SQL Server társított szolgáltatásának létrehozása

Ebben a lépésben az SQL Server-adatbázist az adat-előállítóhoz csatolja.

  1. A Connections (Kapcsolatok) ablakban váltson az Integration Runtimes (Integrációs modulok) lapról a Linked Services (Társított szolgáltatások) lapra, és kattintson a + New (+ Új) elemre.

    New linked service.

  2. A New Linked Service (Új társított szolgáltatás) ablakban válassza az SQL Server lehetőséget, majd kattintson a Continue (Folytatás) gombra.

  3. Az Új társított szolgáltatás ablakban végezze el az alábbi lépéseket:

    1. A Név mezőben adja meg az SqlServerLinkedService értéket.
    2. Válassza a MySelfHostedIR elemet a Connect via integration runtime (Csatlakozás integrációs modulon keresztül) lehetőségnél. Ez egy fontos lépés. Az alapértelmezett integrációs modul nem tud csatlakozni a helyszíni adattárakhoz. Használja a korábban létrehozott saját üzemeltetésű integrációs modult.
    3. A Kiszolgáló neve mezőben adja meg az SQL Server-adatbázist tartalmazó számítógép nevét.
    4. Az Adatbázis neve mezőben adja meg a forrásadatokkal rendelkező SQL Server-adatbázis nevét. Az előfeltételek részeként létrehozott egy táblát, és adatokat szúrt be ebbe az adatbázisba.
    5. Az Authentication type(Hitelesítési típus) mezőben válassza ki a hitelesítés típusát, amellyel az adatbázishoz kíván csatlakozni.
    6. A User name (Felhasználónév) mezőben adja meg az SQL Server-adatbázishoz hozzáféréssel rendelkező felhasználó nevét. Ha perjel karaktert (\) kell használnia a felhasználói fiók vagy a kiszolgáló nevében, használja az escape-karaktert (\). Például: mydomain\\myuser.
    7. A Password (Jelszó) mezőben adja meg a felhasználónévhez tartozó jelszót.
    8. Ha tesztelni szeretné, hogy a Data Factory csatlakozni tud-e az SQL Server-adatbázishoz, kattintson a Test connection (Kapcsolat tesztelése) gombra. Javítson ki minden hibát, amíg nem sikerül a kapcsolódás.
    9. A társított szolgáltatás mentéséhez kattintson a Befejezés gombra.

Az Azure SQL Database társított szolgáltatás létrehozása

Az utolsó lépésben létrehoz egy társított szolgáltatást, amely összekapcsolja az SQL Server-adatbázist az adat-előállítóval. Ebben a lépésben összekapcsolja a cél-/fogadó-adatbázist az adat-előállítóval.

  1. A Connections (Kapcsolatok) ablakban váltson az Integration Runtimes (Integrációs modulok) lapról a Linked Services (Társított szolgáltatások) lapra, és kattintson a + New (+ Új) elemre.

  2. Az Új társított szolgáltatás ablakban válassza az Azure SQL Database lehetőséget, majd kattintson a Folytatás elemre.

  3. Az Új társított szolgáltatás ablakban végezze el az alábbi lépéseket:

    1. A Név mezőbe írja az AzureSqlDatabaseLinkedService nevet.
    2. Kiszolgálónév esetén válassza ki a kiszolgáló nevét a legördülő listából.
    3. Az adatbázis neveként válassza ki azt az adatbázist, amelyben customer_table létrehozott, és project_table az előfeltételek részeként.
    4. Felhasználónévként adja meg az adatbázishoz hozzáféréssel rendelkező felhasználó nevét.
    5. A Password (Jelszó) mezőben adja meg a felhasználónévhez tartozó jelszót.
    6. Ha tesztelni szeretné, hogy a Data Factory csatlakozni tud-e az SQL Server-adatbázishoz, kattintson a Test connection (Kapcsolat tesztelése) gombra. Javítson ki minden hibát, amíg nem sikerül a kapcsolódás.
    7. A társított szolgáltatás mentéséhez kattintson a Befejezés gombra.

  4. Győződjön meg róla, hogy a két társított szolgáltatás szerepel a listában.

    Two linked services

Adatkészletek létrehozása

Ebben a lépésben olyan adatkészleteket hoz létre, amelyek az adatforrást, az adat célhelyét és a küszöbérték tárolási helyét jelölik.

Forrásadatkészlet létrehozása

  1. Kattintson a bal oldali ablaktáblán a + (plusz) jelre, majd a Dataset (Adatkészlet) elemre.

  2. Az Új adathalmaz ablakban válassza az SQL Servert, és kattintson a Folytatás gombra.

  3. A webböngészőben megjelenik egy új lap, amely az adatkészlet konfigurálására szolgál. Egy adathalmaz is megjelenik a fa nézetben. A Properties (Tulajdonságok) ablak General (Általános) lapjának alján a SourceDataset értéket adja meg a Name (Név) mezőben.

  4. Váltson a Connection (Kapcsolat) lapra a Tulajdonságok ablakban, majd a Linked service (Társított szolgáltatás) mezőben válassza az SqlServerLinkedService elemet. Itt ne válasszon táblát. A teljes tábla betöltése helyett a folyamat másolási tevékenysége egy SQL-lekérdezést használ az adatok betöltéséhez.

    Source dataset - connection

Fogadó adatkészlet létrehozása

  1. Kattintson a bal oldali ablaktáblán a + (plusz) jelre, majd a Dataset (Adatkészlet) elemre.

  2. Az Új adatkészlet ablakban válassza az Azure SQL Database lehetőséget, és kattintson a Folytatás gombra.

  3. A webböngészőben megjelenik egy új lap, amely az adatkészlet konfigurálására szolgál. Egy adathalmaz is megjelenik a fa nézetben. A Properties (Tulajdonságok) ablak General (Általános) lapjának alján a SinkDataset értéket adja meg a Name (Név) mezőben.

  4. Váltson a Parameters (Paraméterek) lapra a tulajdonságok ablakában, és hajtsa végre a következő lépéseket:

    1. Kattintson a New (Új) elemre a Create/update parameters (Paraméterek létrehozása/frissítése) szakaszban.

    2. Adja meg a SinkTableNamenevet és a String (Sztring) típust. Ez az adatkészlet a SinkTableName paramétert használja. A SinkTableName paramétert a folyamat állítja be dinamikusan, futásidőben. A folyamat ForEach tevékenysége végighalad a táblanevek listáján, és minden egyes ismétléskor átadja a táblanevet ennek az adatkészletnek.

      Sink Dataset - properties

  5. Váltson a Tulajdonságok ablak Csatlakozás ion lapjára, és válassza az AzureSqlDatabaseLinkedService for Linked service lehetőséget. A Table (Tábla) tulajdonsághoz kattintson az Add dynamic content (Dinamikus tartalom hozzáadása) lehetőségre.

  6. A Dinamikus tartalom hozzáadása ablakban válassza a SinkTableName lehetőséget a Paraméterek szakaszban.

  7. A Befejezés gombra kattintva megjelenik a "@dataset(). SinkTableName" táblanévként.

    Sink Dataset - connection

Adatkészlet létrehozása a küszöbhöz

Ebben a lépésben egy adatkészletet hozunk létre a felső küszöbértékek tárolására.

  1. Kattintson a bal oldali ablaktáblán a + (plusz) jelre, majd a Dataset (Adatkészlet) elemre.

  2. Az Új adatkészlet ablakban válassza az Azure SQL Database lehetőséget, és kattintson a Folytatás gombra.

  3. A Properties (Tulajdonságok) ablak General (Általános) lapjának alján a WatermarkDataset értéket adja meg a Name (Név) mezőben.

  4. Váltson a Kapcsolat lapra, és végezze el az alábbi lépéseket:

    1. A Társított szolgáltatás elemnél válassza az AzureSqlDatabaseLinkedService lehetőséget.

    2. A Tábla mezőnél válassza a [dbo].[watermarktable] lehetőséget.

      Watermark Dataset - connection

Folyamat létrehozása

A folyamat táblanevek listáját használja paraméterként. A ForEach tevékenység végighalad a táblanevek listáján, és elvégzi a következő műveleteket:

  1. A keresési tevékenység lekéri a régi küszöbértéket (a kezdeti értéket, vagy a legutóbbi ismétlés során használt értéket).

  2. A keresési tevékenység lekéri az új küszöbértéket (a forrástábla küszöbértéket tartalmazó oszlopának legnagyobb értékét).

  3. A másolási tevékenység az előbbi két küszöbérték közötti adatokat másolja a forrásadatbázisból a céladatbázisba.

  4. A StoredProcedure (tárolt eljárás) tevékenység frissíti a régi küszöbértéket, hogy a következő ismétlés első lépése azt használja.

A folyamat létrehozása

  1. Kattintson a bal oldali ablaktáblán a + (plusz) jelre, majd kattintson a Folyamat elemre.

  2. Az Általános panel Tulajdonságok területén adja meg a Név növekményesCopyPipeline elemét. Ezután a jobb felső sarokban található Tulajdonságok ikonra kattintva összecsukja a panelt.

  3. A Paraméterek lapon hajtsa végre a következő lépéseket:

    1. Kattintson az + Új elemre.
    2. A paraméter nevénél adja meg a tableList nevet.
    3. Válassza a Tömb lehetőséget a paramétertípushoz.

  4. A tevékenységek eszközkészletében bontsa ki az Ismétlés és feltételek elemet, és húzza a ForEach tevékenységet a folyamat tervezőfelületére. A Properties (Tulajdonságok) ablak General (Általános) lapján az IterateSQLTables értéket adja meg.

  5. Váltson a Beállítások lapra, majd az Elemek értékeként adja meg a következőt: @pipeline().parameters.tableList. A ForEach tevékenység végighalad egy táblalistán, és végrehajtja a növekményes másolási műveletet.

    ForEach activity - settings

  6. Válassza ki a ForEach tevékenységet a folyamatban, ha még nincs kiválasztva. Kattintson az Edit (Szerkesztés) gombra (ceruza ikon).

  7. A tevékenységek eszközkészletében bontsa ki az Általános elemet, húzza a keresési tevékenységet a folyamat tervezőfelületére, és a Név mezőbe írja be a LookupOldWaterMarkActivity kifejezést.

  8. Váltson a Beállítások lapra a tulajdonságok ablakában, és hajtsa végre a következő lépéseket:

    1. Válassza ki a WatermarkDataset elemet a Source Dataset (Forrásadatkészlet) mezőben.

    2. A Lekérdezés használata elemnél válassza a Lekérdezés lehetőséget.

    3. A Lekérdezés elemhez adja meg az alábbi SQL-lekérdezést.

      select * from watermarktable where TableName  =  '@{item().TABLE_NAME}'

      First Lookup Activity - settings

  9. Húzza a Lookup (Keresés) tevékenységet az Activities (Tevékenységek) eszközkészletből, és írja be a LookupNewWaterMarkActivitynevet.

  10. Váltson a Settings (Beállítások) lapra.

    1. Válassza ki a SourceDataset elemet a Forrásadatkészlet mezőnél.

    2. A Lekérdezés használata elemnél válassza a Lekérdezés lehetőséget.

    3. A Lekérdezés elemhez adja meg az alábbi SQL-lekérdezést.

      select MAX(@{item().WaterMark_Column}) as NewWatermarkvalue from @{item().TABLE_NAME}

      Second Lookup Activity - settings

  11. Húzza a Copy (Másolás) tevékenységet az Activities (Tevékenységek) eszközkészletből, és írja be az IncrementalCopyActivitynevet.

  12. Kapcsolja egymás után a Lookup (Keresés) tevékenységeket a Copy (Másolás) tevékenységhez. A csatlakozáshoz húzza a Lookup (Keresés) tevékenységhez tartozó zöld gombot a Copy (Másolás) tevékenységre. Amikor a másolási tevékenység szegélyének színe kékre vált, engedje el az egér gombját.

    Connect Lookup activities to Copy activity

  13. Válassza ki a folyamatban a Copy (Másolás) tevékenységet. Váltson a Source (Forrás) lapra a Properties (Tulajdonságok) ablakban.

    1. Válassza ki a SourceDataset elemet a Forrásadatkészlet mezőnél.

    2. A Lekérdezés használata elemnél válassza a Lekérdezés lehetőséget.

    3. A Lekérdezés elemhez adja meg az alábbi SQL-lekérdezést.

      select * from @{item().TABLE_NAME} where @{item().WaterMark_Column} > '@{activity('LookupOldWaterMarkActivity').output.firstRow.WatermarkValue}' and @{item().WaterMark_Column} <= '@{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}'

      Copy Activity - source settings

  14. Váltson a Sink (Fogadó) lapra, és válassza a SinkDataset lehetőséget a Sink Dataset (Fogadó adatkészlet) mezőnél.

  15. Hajtsa végre a következő lépéseket:

    1. Az Adathalmaz tulajdonságai között adja meg a SinkTableName paramétert@{item().TABLE_NAME}.

    2. A Tárolt eljárás neve tulajdonsághoz írja be a következőt @{item().StoredProcedureNameForMergeOperation}:

    3. A Táblatípus tulajdonsághoz írja be a következőt@{item().TableType}:

    4. A Táblatípus paraméter neve mezőbe írja be a következőt @{item().TABLE_NAME}:

      Copy Activity - parameters

  16. Húzza át a Tárolt eljárás tevékenységet a tevékenységek eszközkészletéből a folyamat tervezőfelületére. Kapcsolja a Copy (Másolás) tevékenységet a Stored Procedure (Tárolt eljárás) tevékenységhez.

  17. Válassza ki a Stored Procedure (Tárolt eljárás) tevékenységet a folyamatban, és írja be a StoredProceduretoWriteWatermarkActivitynevet a Properties (Tulajdonságok) ablak General (Általános) lapján.

  18. Váltson az SQL Account (SQL-fiók) lapra, majd a Linked Service (Társított szolgáltatás) elemnél válassza az AzureSqlDatabaseLinkedService lehetőséget.

    Stored Procedure Activity - SQL Account

  19. Váltson a Tárolt eljárás lapra, és végezze el az alábbi lépéseket:

    1. A tárolt eljárás neveként válassza az [dbo].[usp_write_watermark] lehetőséget.

    2. Válassza az Importálási paraméter lehetőséget.

    3. Adja meg a következő értékeket a paraméterekhez:

      Name Típus Érték
      LastModifiedtime DateTime @{activity('LookupNewWaterMarkActivity').output.firstRow.NewWatermarkvalue}
      TableName Sztring @{activity('LookupOldWaterMarkActivity').output.firstRow.TableName}

      Stored Procedure Activity - stored procedure settings

  20. A Data Factory szolgáltatásban létrehozott entitások közzétételéhez válassza az Összes közzététele lehetőséget.

  21. Várjon, amíg megjelenik a Sikeres közzététel üzenet. Az értesítések megtekintéséhez kattintson a Show Notifications (Értesítések megjelenítése) hivatkozásra. Az X gombra kattintva zárja be az értesítések ablakát.

A folyamat futtatása

  1. A folyamat eszköztárán kattintson az Eseményindító hozzáadása, majd az Eseményindító most gombra.

  2. A Pipeline Run (Folyamatfuttatás) ablakban írja be a következő értéket a tableList paraméterhez, és kattintson a Finish (Befejezés) gombra.

    [
    {
        "TABLE_NAME": "customer_table",
        "WaterMark_Column": "LastModifytime",
        "TableType": "DataTypeforCustomerTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_customer_table"
    },
    {
        "TABLE_NAME": "project_table",
        "WaterMark_Column": "Creationtime",
        "TableType": "DataTypeforProjectTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_project_table"
    }
]

    Pipeline Run arguments

A folyamat figyelése

  1. Váltson a bal oldali Monitorozás lapra. Láthatja a manuális eseményindítás által elindított folyamatfuttatást. A FOLYAMATNÉV oszlop hivatkozásai segítségével megtekintheti a tevékenység részleteit, és újrafuttathatja a folyamatot.

  2. A folyamatfuttatáshoz társított tevékenységfuttatások megtekintéséhez válassza a hivatkozást a FOLYAMATNÉV oszlop alatt. A tevékenység futtatásával kapcsolatos részletekért válassza a Részletek hivatkozást (szemüveg ikon) a TEVÉKENYSÉGNÉV oszlopban.

  3. A folyamatfuttatások nézetre való visszalépéshez válassza a felül található Összes folyamatfuttatás lehetőséget. A nézet frissítéséhez válassza a Frissítés parancsot.

Az eredmények áttekintése

Az SQL Server Management Studióban futtassa a következő lekérdezéseket a cél SQL-adatbázison annak ellenőrzéséhez, hogy a rendszer átmásolta-e az adatokat a forrástáblákból a céltáblákba:

Lekérdezés

select * from customer_table

Hozam

===========================================
PersonID	Name	LastModifytime
===========================================
1	        John	2017-09-01 00:56:00.000
2	        Mike	2017-09-02 05:23:00.000
3	        Alice	2017-09-03 02:36:00.000
4	        Andy	2017-09-04 03:21:00.000
5	        Anny	2017-09-05 08:06:00.000

Lekérdezés

select * from project_table

Hozam

===================================
Project	    Creationtime
===================================
project1	2015-01-01 00:00:00.000
project2	2016-02-02 01:23:00.000
project3	2017-03-04 05:16:00.000

Lekérdezés

select * from watermarktable

Hozam

======================================
TableName	    WatermarkValue
======================================
customer_table	2017-09-05 08:06:00.000
project_table	2017-03-04 05:16:00.000

Megfigyelhető, hogy mindkét tábla küszöbértékei frissültek.

További adatok hozzáadása a forrástáblákhoz

Futtassa a következő lekérdezést a forrás SQL Server-adatbázison a customer_table meglévő sorának frissítéséhez. Szúrjon be egy új sort a project_table táblába.

UPDATE customer_table
SET [LastModifytime] = '2017-09-08T00:00:00Z', [name]='NewName' where [PersonID] = 3

INSERT INTO project_table
(Project, Creationtime)
VALUES
('NewProject','10/1/2017 0:00:00 AM');

A folyamat újbóli futtatása

  1. A webböngésző ablakában váltson a bal oldalon található Edit (Szerkesztés) lapra.

  2. A folyamat eszköztárán kattintson az Eseményindító hozzáadása, majd az Eseményindító most gombra.

  3. A Pipeline Run (Folyamatfuttatás) ablakban írja be a következő értéket a tableList paraméterhez, és kattintson a Finish (Befejezés) gombra.

    [
    {
        "TABLE_NAME": "customer_table",
        "WaterMark_Column": "LastModifytime",
        "TableType": "DataTypeforCustomerTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_customer_table"
    },
    {
        "TABLE_NAME": "project_table",
        "WaterMark_Column": "Creationtime",
        "TableType": "DataTypeforProjectTable",
        "StoredProcedureNameForMergeOperation": "usp_upsert_project_table"
    }
]

A folyamat ismételt monitorozása

  1. Váltson a bal oldali Monitorozás lapra. Láthatja a manuális eseményindítás által elindított folyamatfuttatást. A FOLYAMATNÉV oszlop hivatkozásai segítségével megtekintheti a tevékenység részleteit, és újrafuttathatja a folyamatot.

  2. A folyamatfuttatáshoz társított tevékenységfuttatások megtekintéséhez válassza a hivatkozást a FOLYAMATNÉV oszlop alatt. A tevékenység futtatásával kapcsolatos részletekért válassza a Részletek hivatkozást (szemüveg ikon) a TEVÉKENYSÉGNÉV oszlopban.

  3. A folyamatfuttatások nézetre való visszalépéshez válassza a felül található Összes folyamatfuttatás lehetőséget. A nézet frissítéséhez válassza a Frissítés parancsot.

A végső eredmények áttekintése

Az SQL Server Management Studióban futtassa a következő lekérdezéseket a cél SQL-adatbázison annak ellenőrzéséhez, hogy a frissített/új adatok át lettek-e másolva a forrástáblákból a céltáblákba.

Lekérdezés

select * from customer_table

Hozam

===========================================
PersonID	Name	LastModifytime
===========================================
1	        John	2017-09-01 00:56:00.000
2	        Mike	2017-09-02 05:23:00.000
3	        NewName	2017-09-08 00:00:00.000
4	        Andy	2017-09-04 03:21:00.000
5	        Anny	2017-09-05 08:06:00.000

A 3. számú PersonID új Name és LastModifytime értékkel rendelkezik.

Lekérdezés

select * from project_table

Hozam

===================================
Project	    Creationtime
===================================
project1	2015-01-01 00:00:00.000
project2	2016-02-02 01:23:00.000
project3	2017-03-04 05:16:00.000
NewProject	2017-10-01 00:00:00.000

A project_table táblához hozzá lett adva a NewProject bejegyzés.

Lekérdezés

select * from watermarktable

Hozam

======================================
TableName	    WatermarkValue
======================================
customer_table	2017-09-08 00:00:00.000
project_table	2017-10-01 00:00:00.000

Megfigyelhető, hogy mindkét tábla küszöbértékei frissültek.

Kapcsolódó tartalom

Az oktatóanyagban az alábbi lépéseket hajtotta végre:

  • Forrás- és céladattárak előkészítése.
  • Adat-előállító létrehozása
  • Helyi Integration Runtime (IR) létrehozása.
  • Az Integration Runtime telepítése.
  • Társított szolgáltatások létrehozása.
  • Forrás-, fogadó- és küszöbadatkészletek létrehozása.
  • Folyamat létrehozása, futtatása és figyelése.
  • Tekintse át az eredményeket.
  • Adatok hozzáadása vagy frissítése a forrástáblákban.
  • A folyamat újrafuttatása és monitorozása.
  • A végső eredmények áttekintése.

Folytassa a következő oktatóanyaggal, amelyben az adatok Azure Spark-fürtök használatával való átalakítását ismerheti meg:

Adatok növekményes betöltése az Azure SQL Database-ből az Azure Blob Storage-ba a változáskövetési technológia használatával