Folyamatok és tevékenységek az Azure Data Factory és Azure Synapse Analyticsben
A KÖVETKEZŐKRE VONATKOZIK:
Azure Data Factory Azure Synapse Analytics
Fontos
A (klasszikus) Machine Learning Studio támogatása 2024. augusztus 31-én megszűnik. Javasoljuk, hogy addig térjen át az Azure Machine Learning használatára.
2021. december 1-től kezdődően nem hozhat létre új (klasszikus) Machine Learning Studio-erőforrásokat (munkaterületi és webszolgáltatási csomagot). 2024. augusztus 31-ig továbbra is használhatja a meglévő (klasszikus) Machine Learning Studio-kísérleteket és -webszolgáltatásokat.
- A gépi tanulási projekteknek a (klasszikus) ML Studióból a Azure Machine Learning való áthelyezésére vonatkozó információk.
- További információ a Azure Machine Learning
A (klasszikus) ML Studio dokumentációjának kivezetése folyamatban van, és a jövőben nem várható a frissítése.
Ez a cikk segítséget nyújt az Azure Data Factory és az Azure Synapse Analytics folyamatainak és tevékenységeinek megértéséhez, valamint az adatok áthelyezéséhez és adatfeldolgozásához használható, végpontok közötti adatvezérelt munkafolyamatok létrehozásához.
Áttekintés
Egy Data Factory vagy Synapse-munkaterület egy vagy több folyamatból is rendelkezhet. A folyamatok olyan tevékenységek logikus csoportosításai, amelyek együttesen vesznek részt egy feladat végrehajtásában. Például a folyamat tartalmazhat egy olyan tevékenységkészletet, amely naplóadatokat tölt be és töröl, majd egy leképezési adatfolyamot futtat a naplóadatok elemzéséhez. A folyamatok lehetővé teszik, hogy a tevékenységeket egy készletben kezelje, ne pedig külön-külön. Magát a folyamatot helyezheti üzembe és ütemezheti az önálló tevékenységek helyett.
A folyamat tevékenységei meghatározzák az adatokon végrehajtandó műveleteket. Másolási tevékenység használatával például adatokat másolhat SQL Server egy Azure Blob Storage. Ezután adatfolyam-tevékenység vagy Databricks Notebook-tevékenység használatával feldolgozhatja és átalakíthatja a blobtároló adatait egy Azure Synapse Analytics-készletté, amelyre üzletiintelligencia-jelentési megoldások épülnek.
Azure Data Factory és Azure Synapse Analytics három tevékenységcsoportból áll: adattovábbítási tevékenységek, adatátalakítási tevékenységek és vezérlési tevékenységek. Egy tevékenység nulla vagy több bemeneti adathalmazt vehet igénybe, és egy vagy több kimeneti adatkészletet hozhat létre. Az alábbi ábrán a folyamat, a tevékenység és az adatkészlet közötti kapcsolat látható:
A bemeneti adatkészlet a folyamatban lévő tevékenység bemenetét, a kimeneti adatkészlet pedig a tevékenység kimenetét jelöli. Az adatkészletek adatokat határoznak meg a különböző adattárakban, például táblákban, fájlokban, mappákban és dokumentumokban. Az adatkészlet létrehozását követően használhatja azt egy folyamat tevékenységei esetében. Az adatkészletek lehetnek például egy másolási tevékenység vagy egy HDInsightHive-tevékenység be- vagy kimeneti adatkészletei. Az adatkészletekről további információkat Az Azure Data Factory adatkészletei cikkben talál.
Adattovábbítási tevékenységek
A Data Factory másolási tevékenysége adatokat másol egy forrásadattárból egy fogadó adattárba. A Data Factory a jelen szakaszban található táblában felsorolt adattárakat támogatja. Az adatok bármilyen forrásból bármilyen fogadóba másolhatók. Az adattárra kattintva megtudhatja, hogy az adott tárolóba, illetve tárolóból hogyan másolhat adatokat.
Megjegyzés
Az előzetes verzió jelzéssel ellátott összekötőket kipróbálhatja, és visszajelzést küldhet róluk. Ha függőséget szeretne felvenni a megoldásában található előzetes verziójú összekötőkre, lépjen kapcsolatba az Azure-támogatással.
További információkat a Másolási tevékenység áttekintése cikkben talál.
Adatátalakítási tevékenységek
Azure Data Factory és Azure Synapse Analytics a következő átalakítási tevékenységeket támogatja, amelyek egyenként vagy egy másik tevékenységgel összefűzve adhatók hozzá.
| Adatátalakítási tevékenység | Számítási környezet |
|---|---|
| Adatfolyam | Azure Data Factory által felügyelt Apache Spark-fürtök |
| Azure-függvény | Azure Functions |
| Hive | HDInsight [Hadoop] |
| Pig | HDInsight [Hadoop] |
| MapReduce | HDInsight [Hadoop] |
| Hadoop Streaming | HDInsight [Hadoop] |
| Spark | HDInsight [Hadoop] |
| ML Studio (klasszikus) tevékenységei: Kötegelt végrehajtás és erőforrás frissítése | Azure VM |
| Tárolt eljárás | Azure SQL, Azure Synapse Analytics vagy SQL Server |
| U-SQL | Azure Data Lake Analytics |
| Egyéni tevékenység | Azure Batch |
| Databricks-jegyzetfüzet | Azure Databricks |
| Databricks Jar-tevékenység | Azure Databricks |
| Databricks Python-tevékenység | Azure Databricks |
További információkért tekintse meg az adatátalakítási tevékenységekről szóló cikket.
Vezérlésfolyam-tevékenységek
A támogatott átvitelvezérlési tevékenységek a következők:
| Vezérlési tevékenység | Description |
|---|---|
| Változó hozzáfűzése | Érték hozzáadása meglévő tömbváltozóhoz. |
| Folyamat végrehajtása | A Folyamat végrehajtása tevékenység lehetővé teszi, hogy egy Data Factory- vagy Synapse-folyamat meghívjon egy másik folyamatot. |
| Szűrő | Szűrőkifejezés alkalmazása bemeneti tömbre |
| Mindegyikhez | A ForEach tevékenység ismétlődő átvitelvezérlést határoz meg a folyamatban. Ez a tevékenység egy gyűjtemény megismétlésére, valamint egy megadott ciklustevékenység végrehajtására szolgál. E tevékenység ciklusos megvalósítása hasonló a Foreach ciklusos szerkezetéhez a programozási nyelvek esetében. |
| Metaadatok lekérése | A GetMetadata tevékenység egy Data Factory- vagy Synapse-folyamat bármely adatának metaadatainak lekérésére használható. |
| If Condition tevékenység | Az If Condition tevékenység igaz vagy hamis értéket visszaadó feltételek alapján történő elágaztatásra használható. Az If Condition tevékenység ugyanazokat a funkciókat biztosítja, mint a programnyelvek if utasítása. Egy tevékenységcsoportot értékel ki, amikor a feltétel kiértékelése befejeződött, true és egy másik tevékenységkészletet, amikor a feltétel kiértékelésének eredménye false. |
| Keresési tevékenység | A Keresési tevékenység segítségével bármely külső forrásból kiolvashat vagy megkereshet egy rekordot, táblanevet vagy értéket. Erre a kimenetre a későbbi tevékenységek is hivatkozhatnak. |
| Változó beállítása | Állítsa be egy meglévő változó értékét. |
| Until tevékenység | A Do-Until ciklus megvalósítása hasonló a programozási nyelvek Do-Until ciklusos szerkezetéhez. Egy tevékenységkészletet futtat le ciklusosan addig, amíg a tevékenységhez rendelt feltétel igaz értéket nem ad vissza. Megadhat egy időtúllépési értéket a until tevékenységhez. |
| Érvényesítési tevékenység | Győződjön meg arról, hogy egy folyamat csak akkor folytatja a végrehajtást, ha létezik referenciaadatkészlet, megfelel egy megadott feltételnek, vagy időtúllépés történt. |
| Wait tevékenység | Ha várakozási tevékenységet használ egy folyamatban, a folyamat a megadott ideig várakozik, mielőtt folytatná a további tevékenységek végrehajtását. |
| Webes tevékenység | A webes tevékenységgel egyéni REST-végpontokat hívhat meg egy folyamatból. Az adatkészleteket és a társított szolgáltatásokat továbbíthatja a tevékenység számára felhasználásra vagy elérés céljára. |
| Webhook tevékenység | A webhook-tevékenység használatával hívjon meg egy végpontot, és adjon át egy visszahívási URL-címet. A folyamatfuttatás megvárja a visszahívás meghívását, mielőtt továbblép a következő tevékenységre. |
Folyamat létrehozása felhasználói felülettel
Új folyamat létrehozásához lépjen a Data Factory Studióban a Szerző lapra (amelyet a ceruza ikon jelöl), majd kattintson a pluszjelre, és válassza a Menü Folyamat elemét, majd az almenüből ismét a Folyamat lehetőséget.
A Data Factory megjeleníti a folyamatszerkesztőt, ahol megtalálja a következőt:
- A folyamaton belül használható összes tevékenység.
- A folyamatszerkesztő vászna, ahol a tevékenységek megjelennek a folyamathoz való hozzáadáskor.
- A folyamatkonfigurációk panel, beleértve a paramétereket, a változókat, az általános beállításokat és a kimenetet.
- A folyamat tulajdonságainak panelje, ahol konfigurálható a folyamat neve, opcionális leírása és széljegyzetei. Ez a panel a folyamathoz kapcsolódó elemeket is megjeleníti az adat-előállítóban.
A folyamat JSON-fájlja
Egy folyamat JSON-formátumban való meghatározása a következő módon történik:
{
"name": "PipelineName",
"properties":
{
"description": "pipeline description",
"activities":
[
],
"parameters": {
},
"concurrency": <your max pipeline concurrency>,
"annotations": [
]
}
}
| Címke | Leírás | Típus | Kötelező |
|---|---|---|---|
| name | A folyamat neve. Adjon meg egy, a folyamat által végrehajtandó műveletet jelölő nevet.
|
Sztring | Yes |
| leírás | Adjon meg egy, az adott folyamat alkalmazását leíró szöveget. | Sztring | No |
| tevékenységek | A tevékenységek szakaszon belül egy vagy több tevékenység is meghatározható. A tevékenységek JSON-elemeiről részletes információkat a Tevékenység JSON-fájlja szakaszban talál. | Tömb | Yes |
| parameters | Az adott folyamat paraméterek szakaszában egy vagy több paraméter adható meg, így a folyamat rugalmasan újrafelhasználható. | Lista | No |
| Konkurencia | A folyamat egyidejű futtatásának maximális száma. Alapértelmezés szerint nincs maximális érték. Ha eléri az egyidejűségi korlátot, a rendszer további folyamatfuttatásokat vár a korábbiak befejezéséig | Szám | No |
| Széljegyzetek | A folyamathoz társított címkék listája | Tömb | No |
Tevékenység JSON-fájlja
A tevékenységek szakaszon belül egy vagy több tevékenység is meghatározható. A következő két fő tevékenységtípust különböztetjük meg: végrehajtási és vezérlési tevékenységek.
Végrehajtási tevékenységek
A végrehajtási tevékenységek közé az adatáthelyezési és az adatátalakítási tevékenységek tartoznak. Ezek a következő felső szintű struktúrával rendelkeznek:
{
"name": "Execution Activity Name",
"description": "description",
"type": "<ActivityType>",
"typeProperties":
{
},
"linkedServiceName": "MyLinkedService",
"policy":
{
},
"dependsOn":
{
}
}
Az alábbi táblában a tevékenység JSON-definíciójában lévő tulajdonságok szerepelnek:
| Címke | Leírás | Kötelező |
|---|---|---|
| name | A tevékenység neve. Adjon meg egy, a tevékenység által végrehajtandó műveletet jelölő nevet.
|
Yes |
| leírás | Az adott tevékenységet vagy annak alkalmazását leíró szöveg | Yes |
| típus | A tevékenység típusa. A különböző típusú tevékenységekről az adatáthelyezési tevékenységek, az adatátalakítási tevékenységek és a vezérlési tevékenységek szakaszában olvashat. | Yes |
| linkedServiceName | A tevékenység által használt társított szolgáltatás neve. Egy adott tevékenység megkövetelheti annak a társított szolgáltatásnak a megadását, amely a szükséges számítási környezethez kapcsolódik. |
Igen a HDInsight-tevékenységhez, ML Studio (klasszikus) kötegelt pontozási tevékenységhez, tárolt eljárástevékenységhez. Minden egyéb esetében: nem |
| typeProperties | A typeProperties szakasz tulajdonságai az egyes tevékenységtípusoktól függenek. Az adott tevékenység típustulajdonságainak megtekintéséhez kattintson az előző szakaszban szereplő tevékenységhivatkozásokra. | No |
| szabályzat | Olyan szabályzatok, amelyek az adott tevékenység futásidejű viselkedését befolyásolják. Ez a tulajdonság időtúllépést és újrapróbálkozások viselkedését tartalmazza. Ha nincs megadva, a rendszer az alapértelmezett értékeket használja. További információkat a Tevékenységszabályzat szakaszban talál. | No |
| dependsOn | Ez a tulajdonság a tevékenységfüggőségek, valamint az egymást követő tevékenységek függőségeinek meghatározására szolgál. További információért lásd: Tevékenységfüggőség | No |
Tevékenységszabályzat
A szabályzatok hatással vannak egy tevékenység futásidejű viselkedésére, és konfigurációs lehetőségeket biztosítanak. A tevékenységszabályzatok kizárólag végrehajtási tevékenységek esetében állnak rendelkezésre.
Tevékenységszabályzat JSON-definíciója
{
"name": "MyPipelineName",
"properties": {
"activities": [
{
"name": "MyCopyBlobtoSqlActivity",
"type": "Copy",
"typeProperties": {
...
},
"policy": {
"timeout": "00:10:00",
"retry": 1,
"retryIntervalInSeconds": 60,
"secureOutput": true
}
}
],
"parameters": {
...
}
}
}
| JSON-név | Description | Megengedett értékek | Kötelező |
|---|---|---|---|
| timeout | Megadja a futtatni kívánt tevékenység időtúllépését. | Időtartomány | Nem. Az alapértelmezett időtúllépés 7 nap. |
| retry | Újrapróbálkozási kísérletek maximális száma | Egész szám | Nem. Az alapértelmezett érték: 0 |
| retryIntervalInSeconds | Az újrapróbálkozási kísérletek közötti késleltetés, másodpercben | Egész szám | Nem. Az alapértelmezett érték 30 másodperc |
| secureOutput | Ha igaz értékre van állítva, a rendszer biztonságosnak tekinti a tevékenység kimenetét, és nem naplózza figyelésre. | Logikai | Nem. Az alapértelmezett érték a false (hamis). |
Vezérlési tevékenység
A vezérlési tevékenységek az alábbi felső szintű struktúrával rendelkeznek:
{
"name": "Control Activity Name",
"description": "description",
"type": "<ActivityType>",
"typeProperties":
{
},
"dependsOn":
{
}
}
| Címke | Leírás | Kötelező |
|---|---|---|
| name | A tevékenység neve. Adjon meg egy, a tevékenység által végrehajtandó műveletet jelölő nevet.
|
Yes |
| leírás | Az adott tevékenységet vagy annak alkalmazását leíró szöveg | Yes |
| típus | A tevékenység típusa. A különböző tevékenységtípusokkal kapcsolatban lásd az adattovábbítási tevékenységeket, az adat-átalakítási tevékenységeket és a vezérlési tevékenységeket. | Yes |
| typeProperties | A typeProperties szakasz tulajdonságai az egyes tevékenységtípusoktól függenek. Az adott tevékenység típustulajdonságainak megtekintéséhez kattintson az előző szakaszban szereplő tevékenységhivatkozásokra. | No |
| dependsOn | Ez a tulajdonság a tevékenységfüggőség, valamint az egymást követő tevékenységek függőségeinek meghatározására szolgál. További információ: tevékenységfüggőség. | No |
Tevékenységfüggőség
A tevékenységfüggőség határozza meg, hogy a további tevékenységek hogyan függenek a korábbi tevékenységektől, és meghatározza, hogy a következő tevékenység végrehajtása folytatódjon-e. Egy adott tevékenység egy vagy több korábbi, eltérő függőségi feltétellel rendelkező tevékenységtől is függhet.
A különböző függőségi feltételek a következők: Sikeres, Sikertelen, Kihagyva, Befejezve.
Ha például egy folyamat rendelkezik A tevékenységgel –> B tevékenységgel, a következő forgatókönyvek fordulhatnak elő:
- A B tevékenység sikeres függőségi feltétellel rendelkezik az A tevékenység esetében: a B tevékenység csak akkor fut le, ha az A tevékenység végállapota sikeres
- A B tevékenység sikertelen függőségi feltétellel rendelkezik az A tevékenység esetében: a B tevékenység csak akkor fut le, ha az A tevékenység végállapota sikertelen
- A B tevékenység befejezve függőségi feltétellel rendelkezik az A tevékenység esetében: a B tevékenység akkor fut le, ha az A tevékenység végállapota sikeres vagy sikertelen
- A B tevékenység függőségi feltétellel rendelkezik a kihagyott A tevékenységhez: A B tevékenység akkor fut, ha az A tevékenység végleges állapota kihagyva. A kihagyott művelet az X – Y tevékenység –>> Z tevékenység forgatókönyvében fordul elő, ahol minden tevékenység csak akkor fut, ha az előző tevékenység sikeres volt. Ha az X tevékenység sikertelen, akkor az Y tevékenység "Kihagyva" állapotú, mert soha nem hajtja végre. Hasonlóképpen a Z tevékenység "Kihagyva" állapotú is.
Példa: a 2. tevékenység az 1. tevékenység sikerességétől függ
{
"name": "PipelineName",
"properties":
{
"description": "pipeline description",
"activities": [
{
"name": "MyFirstActivity",
"type": "Copy",
"typeProperties": {
},
"linkedServiceName": {
}
},
{
"name": "MySecondActivity",
"type": "Copy",
"typeProperties": {
},
"linkedServiceName": {
},
"dependsOn": [
{
"activity": "MyFirstActivity",
"dependencyConditions": [
"Succeeded"
]
}
]
}
],
"parameters": {
}
}
}
Minta másolási folyamat
Az alábbi mintafolyamat tevékenységek szakaszában egyetlen Másolás típusú tevékenység található. Ebben a példában a másolási tevékenység adatokat másol egy Azure Blob Storage-ból egy Azure SQL Database-adatbázisba.
{
"name": "CopyPipeline",
"properties": {
"description": "Copy data from a blob to Azure SQL table",
"activities": [
{
"name": "CopyFromBlobToSQL",
"type": "Copy",
"inputs": [
{
"name": "InputDataset"
}
],
"outputs": [
{
"name": "OutputDataset"
}
],
"typeProperties": {
"source": {
"type": "BlobSource"
},
"sink": {
"type": "SqlSink",
"writeBatchSize": 10000,
"writeBatchTimeout": "60:00:00"
}
},
"policy": {
"retry": 2,
"timeout": "01:00:00"
}
}
]
}
}
Vegye figyelembe a következő szempontokat:
- A tevékenységek szakaszban csak egyetlen tevékenység van, amelynek a típusaCopy értékre van beállítva.
- A tevékenység bemenetének beállítása InputDataset, a kimeneté pedig OutputDataset. Az adatkészletek JSON-fáljban történő meghatározását lásd az Adatkészletek cikket.
- A typeProperties szakaszban forrástípusként a BlobSource, fogadótípusként pedig az SqlSink érték van megadva. Az adattovábbítási tevékenységek szakaszban kattintson a forrásként vagy fogadóként használni kívánt adattárra, hogy további információkhoz jusson az adott adattár esetén a kifelé vagy befelé irányuló adatáthelyezési lehetőségekről.
A folyamat létrehozásának teljes útmutatójáért tekintse meg a Data Factory létrehozását ismertető rövid útmutatót.
Minta átalakítási folyamat
Az alábbi mintafolyamat tevékenységek szakaszában egyetlen HDInsightHive típusú tevékenység található. Ebben a mintában a HDInsight Hive-tevékenység egy Azure blobtárolóból származó adatokat alakít át egy Hive-szkriptfájl Azure HDInsight Hadoop-fürtön történő futtatásával.
{
"name": "TransformPipeline",
"properties": {
"description": "My first Azure Data Factory pipeline",
"activities": [
{
"type": "HDInsightHive",
"typeProperties": {
"scriptPath": "adfgetstarted/script/partitionweblogs.hql",
"scriptLinkedService": "AzureStorageLinkedService",
"defines": {
"inputtable": "wasb://adfgetstarted@<storageaccountname>.blob.core.windows.net/inputdata",
"partitionedtable": "wasb://adfgetstarted@<storageaccountname>.blob.core.windows.net/partitioneddata"
}
},
"inputs": [
{
"name": "AzureBlobInput"
}
],
"outputs": [
{
"name": "AzureBlobOutput"
}
],
"policy": {
"retry": 3
},
"name": "RunSampleHiveActivity",
"linkedServiceName": "HDInsightOnDemandLinkedService"
}
]
}
}
Vegye figyelembe a következő szempontokat:
- A tevékenységek szakaszban csak egyetlen tevékenység van, amelynek a típusaHDInsightHive értékre van beállítva.
- A Hive-szkriptfájl( partitionweblogs.hql) az Azure Storage-fiókban (az AzureStorageLinkedService nevű scriptLinkedService által meghatározott) és a tároló
adfgetstartedszkriptmappájában található. - A
definesszakasz meghatározza a futásidő beállításait, amelyek Hive konfigurációs értékekként (például ${hiveconf:inputtable},${hiveconf:partitionedtable}) lesznek átadva a Hive-parancsfájlnak.
A typeProperties szakasz eltérő az egyes átalakítási tevékenységek esetében. Ahhoz, hogy megismerkedhessen az egyes átalakítási tevékenységek által támogatott típustulajdonságokkal, kattintson az adott átalakítási tevékenységre az Adatátalakítási tevékenységek szakaszban.
E folyamat létrehozásának teljes leírását lásd: Oktatóanyag: adatátalakítás a Spark használatával.
Több tevékenység egy adott folyamatban
Az előző két mintában a folyamatok csak egyetlen tevékenységet tartalmaztak. Egy folyamathoz azonban több tevékenység is tartozhat. Ha az adott folyamatban több tevékenység is található, és az egymást követő tevékenységek nem függnek az azokat megelőző tevékenységektől, akkor ezek párhuzamosan is futtathatók.
A tevékenységfüggőség segítségével összefűzhet két tevékenységet. Ez a fajta függőség azt határozza meg, hogy az egymást követő tevékenységek milyen függőségi viszonyban vannak a megelőző tevékenységekkel, meghatározva azt a feltételt, amelytől a következő feladat végrehajtása függ. Egy adott tevékenység egy vagy több korábbi, eltérő függőségi feltétellel rendelkező tevékenységtől is függhet.
Folyamatok ütemezése
A folyamatok ütemezése eseményindítókkal történik. Különböző típusú eseményindítók léteznek (ütemező eseményindító, amely lehetővé teszi a folyamatok indítását óraalapú ütemezés szerint, valamint a manuális eseményindító, amely igény szerint aktiválja a folyamatokat). Az eseményindítókkal kapcsolatos további információkért tekintse meg a folyamat végrehajtását és az eseményindítókat ismertető cikket.
Ahhoz, hogy az eseményindító kiváltsa egy folyamat indítását, az eseményindító meghatározásába bele kell foglalni az adott folyamat referenciáját. A folyamatok eseményindítói & n-m kapcsolatokkal rendelkeznek. Egyetlen folyamatot több eseményindító is indíthat, és ugyanaz az eseményindító indíthat több folyamatot is. Ha az eseményindító meghatározása megtörtént, el kell indítania azt, hogy az képes legyen az adott folyamat indítására. Az eseményindítókkal kapcsolatos további információkért tekintse meg a folyamat végrehajtását és az eseményindítókat ismertető cikket.
Tegyük fel például, hogy van egy "Trigger A" nevű ütemező eseményindítója, amelyet szeretnék elindítani a "MyCopyPipeline" folyamatból. Az eseményindítót az alábbi példában látható módon határozhatja meg:
Az A eseményindító meghatározása
{
"name": "TriggerA",
"properties": {
"type": "ScheduleTrigger",
"typeProperties": {
...
}
},
"pipeline": {
"pipelineReference": {
"type": "PipelineReference",
"referenceName": "MyCopyPipeline"
},
"parameters": {
"copySourceName": "FileSource"
}
}
}
}
Következő lépések
A folyamatok tevékenységekkel együtt történő létrehozásáról részletes útmutatást a következő oktatóanyagokban talál:
- Másolási tevékenységgel rendelkező folyamat létrehozása
- Adatátalakítási tevékenységgel rendelkező folyamat létrehozása
CI/CD (folyamatos integráció és teljesítés) elérése Azure Data Factory használatával