Pipelines a aktivity v Azure Data Factory a Azure Synapse Analytics

PLATÍ PRO: Azure Data Factory Azure Synapse Analytics

Tento článek vám pomůže pochopit kanály a aktivity ve službě Azure Data Factory a Azure synapse Analytics a použít je k sestavení kompletních pracovních postupů založených na datech pro vaše scénáře přesunu dat a zpracování dat.

Přehled

Pracovní prostor Data Factory nebo synapse může obsahovat jeden nebo více kanálů. Kanál je logické seskupení aktivit, které společně provádějí úlohu. Kanál může například obsahovat sadu aktivit, které ingestují a čistí data protokolu a pak odstartují mapování toku dat k analýze data protokolu. Kanál umožňuje spravovat aktivity jako sadu, místo toho, abyste je museli spravovat jednotlivě. Místo toho, abyste nezávisle nasazovali a plánovali aktivity, nasadíte a naplánujete kanál.

Aktivity v kanálu definují akce, které se mají provádět s daty. aktivitu kopírování můžete například použít ke kopírování dat z SQL Server do Azure Blob Storage. Pak použijte aktivitu toku dat nebo aktivitu poznámkového bloku datacihly ke zpracování a transformaci dat z úložiště objektů blob do fondu služby Azure synapse Analytics nad tím, které business intelligence řešení pro vytváření sestav sestavená.

Azure Data Factory a Azure synapse Analytics mají tři skupiny aktivit: aktivity přesunu dat, aktivity transformace data aktivity řízení. Aktivita může mít nula nebo více vstupních datových sad a vytvoří jednu nebo více výstupních datových sad. Následující diagram znázorňuje vztah mezi kanálem, aktivitou a datovou sadou:

Vstupní datová sada představuje vstup pro aktivitu v kanálu a výstupní datová sada představuje výstup aktivity. Datové sady identifikují data v rámci různých úložišť dat, jako jsou tabulky, soubory, složky a dokumenty. Po vytvoření datové sady můžete tuto datovou sadu používat v aktivitách v rámci kanálu. Datová sada například může být vstupní/výstupní datovou sadou aktivity kopírování nebo aktivity HDInsightHive. Další informace o datových sadách najdete v článku Datové sady v Azure Data Factory.

Aktivity přesunu dat

Aktivita kopírování ve službě Data Factory kopíruje data ze zdrojového úložiště dat do úložiště dat jímky. Služba Data Factory podporuje úložiště dat uvedená v tabulce v této části. Data z libovolného zdroje lze zapsat do libovolné jímky. Kliknutím na úložiště dat se dozvíte, jak kopírovat data z a do daného úložiště.

Další informace najdete v článku Aktivita kopírování – přehled.

Aktivity transformace dat

Azure Data Factory a Azure synapse Analytics podporují následující aktivity transformace, které je možné přidat jednotlivě nebo zřetězit s jinou aktivitou.

Další informace najdete v článku Aktivity transformace dat.

Aktivity toku řízení

Podporují se následující aktivity toku řízení:

Zápis JSON kanálu

Tady je způsob definice kanálu ve formátu JSON:

{
    "name": "PipelineName",
    "properties":
    {
        "description": "pipeline description",
        "activities":
        [
        ],
        "parameters": {
        },
        "concurrency": <your max pipeline concurrency>,
        "annotations": [
        ]
    }
}

Zápis JSON aktivity

Část activities může obsahovat definici jedné nebo více aktivit. Existují dva hlavní typy aktivit: aktivity spuštění a aktivity řízení.

Aktivity spuštění

Aktivity spuštění zahrnují aktivity přesunu dat a transformace dat. Mají následující strukturu nejvyšší úrovně:

{
    "name": "Execution Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "linkedServiceName": "MyLinkedService",
    "policy":
    {
    },
    "dependsOn":
    {
    }
}

Následující tabulka obsahuje popis vlastností v definici aktivity ve formátu JSON:

Zásada aktivity

Zásady ovlivňují chování aktivity za běhu a nabízejí možnosti konfigurace. Zásady aktivit jsou dostupné jenom pro aktivity spuštění.

Definice zásady aktivity ve formátu JSON

{
    "name": "MyPipelineName",
    "properties": {
      "activities": [
        {
          "name": "MyCopyBlobtoSqlActivity",
          "type": "Copy",
          "typeProperties": {
            ...
          },
         "policy": {
            "timeout": "00:10:00",
            "retry": 1,
            "retryIntervalInSeconds": 60,
            "secureOutput": true
         }
        }
      ],
        "parameters": {
           ...
        }
    }
}

Aktivita řízení

Aktivity řízení mají následující strukturu nejvyšší úrovně:

{
    "name": "Control Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "dependsOn":
    {
    }
}

Závislost aktivit

Závislost aktivity definuje, jak následující aktivity závisejí na předchozích aktivitách, a určuje podmínku, jestli se má pokračovat v provádění další úlohy. Aktivita může záviset na jedné nebo více předchozích aktivitách s různými podmínkami závislosti.

Existují různé podmínky závislosti: Úspěch, Chyba, Vynecháno, Dokončeno.

Pokud má například kanál scénář aktivita A -> aktivita B, mohou nastat různé scénáře:

• Aktivita B má podmínku závislosti na aktivitě A s hodnotou Úspěch: aktivita B se spustí jenom v případě, že má aktivita A konečný stav Úspěch.
• Aktivita B má podmínku závislosti na aktivitě A s hodnotou Chyba: aktivita B se spustí jenom v případě, že má aktivita A konečný stav Chyba.
• Aktivita B má podmínku závislosti na aktivitě A s hodnotou Dokončeno: aktivita B se spustí v případě, že má aktivita A konečný stav Dokončeno.
• Aktivita B má podmínku závislosti na aktivitě A s vynechání: Aktivita B se spustí, pokud má aktivita A konečný stav Vynecháno. Stav Přeskočeno nastává ve scénáři aktivita X -> aktivita Y -> aktivita Z, kde se každá aktivita spustí jenom v případě, že úspěšně proběhnou předchozí aktivity. Pokud aktivita X selže, aktivita Y je ve stavu Vynecháno, protože se nikdy neprovede. Podobně má aktivita Z také stav Vynecháno.

Příklad: Aktivita 2 závisí na předchozí aktivitě 1.

{
    "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": {
       }
    }
}

Ukázkový kanál kopírování

V následujícím ukázkovém kanálu je v části activities jedna aktivita typu Kopírování. V této ukázce aktivita kopírování kopíruje data z úložiště objektů blob v Azure do databáze v Azure SQL Database.

{
  "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"
        }
      }
    ]
  }
}

Je třeba počítat s následujícím:

• V části aktivit je jenom jedna aktivita, jejíž vlastnost type je nastavená na Copy.
• Vstup aktivity je nastavený na InputDataset a výstup aktivity je nastavený na OutputDataset. Informace o definicích datových sad ve formátu JSON najdete v článku Datové sady.
• V části typeProperties je jako typ zdroje určen BlobSource a jako typ jímky SqlSink. Pokud chcete zjistit více informací o přesouvání dat do úložiště dat a z něj, v části Aktivity přesunu dat klikněte na úložiště dat, které chcete použít jako zdroj nebo jímku.

Úplný návod k vytvoření tohoto kanálu najdete v tématu Rychlý start: vytvoření Data Factory.

Ukázkový kanál transformace

V následujícím ukázkovém kanálu je v části activities jedna aktivita typu HDInsightHive. V této ukázce aktivita HDInsight Hive transformuje data ze služby Azure Blob Storage tak, že v clusteru Azure HDInsight Hadoop spustí soubor skriptu Hive.

{
    "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"
            }
        ]
    }
}

Je třeba počítat s následujícím:

• V části activities je jenom jedna aktivita, jejíž vlastnost type má hodnotu HDInsightHive.
• Soubor skriptu Hive partitionweblogs.hql je uložený v účtu Azure Storage (který specifikuje služba scriptLinkedService s názvem AzureStorageLinkedService) a ve složce script v kontejneru adfgetstarted .
• Část defines určuje nastavení běhového prostředí, které se předá skriptu Hive jako konfigurační hodnoty Hive (např. ${hiveconf:inputtable}, ${hiveconf:partitionedtable}).

Část typeProperties je u každé aktivity transformace odlišná. Další informace o vlastnostech typu podporovaných u aktivit transformace získáte kliknutím na aktivitu transformace v části Aktivity transformace dat.

Kompletní postup vytváření tohoto kanálu najdete v části Kurz: Transformace dat pomocí jazyka Spark.

Více aktivit v kanálu

Oba předchozí ukázkové kanály obsahovaly jenom jednu aktivitu. Kanál může obsahovat víc než jednu aktivitu. Pokud máte kanál s více aktivitami a následující aktivity nejsou závislé na předchozích aktivitách, mohou se aktivity spouštět souběžně.

Dvě aktivity můžete zřetězit pomocí závislosti aktivit, která určuje, jak následující aktivity závisejí na předchozích aktivitách, a stanovuje podmínku určující, jestli se má pokračovat provedením další úlohy. Aktivita může záviset na jedné nebo více předchozích aktivitách s různými podmínkami závislosti.

Plánování kanálů

Kanály se plánují pomocí aktivačních událostí. Existují různé typy triggerů (aktivační událost plánovače, která umožňuje spouštění kanálů podle hodinového plánu, a také ruční aktivační událost, která aktivuje kanály na vyžádání). Další informace o aktivačních událostech najdete v článku Spouštění kanálů a aktivační události.

Pokud chcete, aby aktivační událost aktivovala spuštění kanálu, musíte do definice aktivační události zahrnout odkaz na příslušný kanál. Mezi kanály a aktivačními událostmi existuje vztah n-m. Více triggerů může aktivovat jeden kanál a jeden trigger může aktivovat více kanálů. Jakmile je aktivační událost definovaná, musíte ji spustit, aby mohla začít aktivovat kanál. Další informace o aktivačních událostech najdete v článku Spouštění kanálů a aktivační události.

Řekněme například, že máte aktivační událost plánovače Trigger A, kterou chci aktivovat kanál MyCopyPipeline. Trigger definujete, jak je znázorněno v následujícím příkladu:

Definice aktivační události Trigger A

{
  "name": "TriggerA",
  "properties": {
    "type": "ScheduleTrigger",
    "typeProperties": {
      ...
      }
    },
    "pipeline": {
      "pipelineReference": {
        "type": "PipelineReference",
        "referenceName": "MyCopyPipeline"
      },
      "parameters": {
        "copySourceName": "FileSource"
      }
    }
  }
}

Další kroky

V následujících kurzech najdete podrobné pokyny pro vytváření kanálů s aktivitami:

• Vytvoření kanálu s aktivitou kopírování
• Vytvoření kanálu s aktivitou transformace dat

Jak dosáhnout CI/CD (kontinuální integrace a doručování) pomocí Azure Data Factory

• Kontinuální integrace a průběžné doručování ve službě Azure Data Factory