Pipelines och aktiviteter i Azure Data Factory och Azure Synapse Analytics

GÄLLER FÖR: Azure Data Factory Azure Synapse Analytics

Den här artikeln hjälper dig att förstå pipelines och aktiviteter i Azure Data Factory och Azure Synapse Analytics och använda dem för att skapa datadrivna arbetsflöden från hela slutet för dina dataförflyttnings- och databearbetningsscenarier.

Översikt

En Data Factory eller Synapse-arbetsyta kan ha en eller flera pipelines. En pipeline är en logisk gruppering av aktiviteter som tillsammans utför en uppgift. En pipeline kan till exempel innehålla en uppsättning aktiviteter som matar in och rensar loggdata, och sedan startar ett mappningsdataflöde för att analysera loggdata. Pipelinen gör att du kan hantera aktiviteterna som en uppsättning i stället för var och en. Du distribuerar och schemalägger pipelinen i stället för aktiviteterna oberoende av varandra.

Aktiviteterna i en pipeline definierar åtgärder som ska utföras på dina data. Du kan till exempel använda en kopieringsaktivitet för att kopiera data från SQL Server till en Azure Blob Storage. Använd sedan en dataflödesaktivitet eller en Databricks Notebook-aktivitet för att bearbeta och transformera data från bloblagringen till en Azure Synapse Analytics-pool där de business intelligence rapportlösningarna har skapats.

Azure Data Factory och Azure Synapse Analytics tre grupperingen av aktiviteter: dataförflyttningsaktiviteter, datatransformeringsaktiviteteroch kontrollaktiviteter. En aktivitet kan ta noll eller flera indatauppsättningar och skapa en eller flera utdatauppsättningar. Följande diagram visar relationen mellan pipeline, aktivitet och datauppsättning:

En indatauppsättning representerar indata för en aktivitet i pipelinen och en utdatauppsättning representerar utdata för aktiviteten. Datauppsättningar identifierar data inom olika datalager, till exempel tabeller, filer, mappar och dokument. När du har skapat en datauppsättning kan du använda den med aktiviteter i en pipeline. Till exempel kan en datauppsättning vara en in-/utdatauppsättning för en kopieringsaktivitet eller en HDInsightHive-aktivitet. Mer information om datauppsättning finns i artikeln Datauppsättningar i Azure Data Factory.

Dataförflyttningsaktiviteter

Kopieringsaktiviteten i Data Factory kopierar data från källans datalager till mottagarens datalager. Data Factory stödjer de datalager som listas i tabellen i det här avsnittet. Data kan skrivas från valfri källa till valfri mottagare. Klicka på ett datalager om du vill veta hur du kopierar data till och från det datalagret.

Mer information finns i artikeln Kopieringsaktiviteten – översikt.

Datatransformeringsaktiviteter

Azure Data Factory och Azure Synapse Analytics stöder följande transformeringsaktiviteter som kan läggas till individuellt eller länkas till en annan aktivitet.

Mer information finns i artikeln om datatransformeringsaktiviteter.

Kontrollflödesaktiviteter

Följande kontrollflödesaktiviteter stöds:

Pipeline JSON

Så här definieras en pipeline i JSON-format:

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

Aktivitets-JSON

Avsnittet activities kan ha en eller flera definierade aktiviteter. Det finns två huvudtyper av aktiviteter: körnings- och kontrollaktiviteter.

Körningsaktiviteter

I körningsaktiviteter ingår dataförflyttning och datatransformering. De har följande toppnivåstruktur:

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

I följande tabell beskrivs egenskaperna i definitionen för aktivitets-JSON:

Aktivitetsprincip

Principer påverkar körningsbeteendet hos en aktivitet och ger konfigurationsalternativ. Aktivitetsprinciper är bar tillgängliga för körningsaktiviteter.

JSON-definition för aktivitetsprincip

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

Kontrollaktivitet

Kontrollaktiviteter har följande toppnivåstruktur:

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

Aktivitetsberoende

Aktivitetsberoende definierar hur efterföljande aktiviteter är beroende av tidigare aktiviteter, vilket fastställer villkoret för om nästa uppgift ska köras. En aktivitet kan vara beroende av en eller flera tidigare aktiviteter med olika beroendevillkor.

De olika beroendevillkoren är Succeeded (Lyckades), Failed (Misslyckades), Skipped (Överhoppad), Completed (Slutförd).

Exempel: Om en pipeline har Aktivitet A -> Aktivitet B är de olika scenarier som kan ske följande:

• Aktivitet B har beroendevillkor på Aktivitet A med Succeeded (Lyckades): Aktivitet B körs bara om Aktivitet A har den slutgiltiga statusen Suceeded (Lyckades)
• Aktivitet B har beroendevillkor på Aktivitet A med Failed (Misslyckades): Aktivitet B körs bara om Aktivitet A har den slutgiltiga statusen Failed (Misslyckades)
• Aktivitet B har beroendevillkor på Aktivitet A med Completed (Slutförd): Aktivitet B körs om Aktivitet A har de slutgiltiga statusen Succeeded (Lyckades) eller Failed (Misslyckades)
• Aktivitet B har ett beroendevillkor på Aktivitet A med överhoppat: Aktivitet B körs om Aktivitet A har den slutgiltiga statusen Skipped (Överhoppad). Skipped (Överhoppad) inträffar i scenariot Aktivitet X -> Aktivitet Y -> Aktivitet Z, där varje aktivitet bara körs om den tidigare aktiviteten lyckas. Om aktivitet X misslyckas har Aktivitet Y statusen "Skipped" (Överhoppad) eftersom den aldrig körs. På samma sätt har aktivitet Z även statusen "Skipped" (Överhoppad).

Exempel: Aktivitet 2 är beroende av att Aktivitet 1 lyckas

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

Exempel på kopieringspipeline

I följande exempel på pipeline finns det en aktivitet av typen Copy (Kopiera) i avsnittet activities. I det här exemplet kopierar kopieringsaktiviteten data från en Azure Blob Storage till en databas i 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"
        }
      }
    ]
  }
}

Observera följande punkter:

• I avsnittet Aktiviteter finns det bara en aktivitet vars typ anges till Kopia.
• Indata för aktiviteten är inställd på InputDataset och utdata för aktiviteten är inställd på OutputDataset. I artikeln Datauppsättningar finns information om hur du definierar datauppsättningar i JSON.
• I avsnittet för typeProperties har BlobSource angetts som källtyp och SqlSink har angetts som mottagartyp. I avsnittet för dataförflyttningsaktiviteter klickar du på det datalager som du vill använda som källa eller mottagare för att lära dig mer om hur du flyttar data till/från det datalagret.

En fullständig genomgång av hur du skapar den här pipelinen finns i Snabbstart: skapa en Data Factory.

Exempel på transfomeringspipeline

I följande exempel på pipeline finns det en aktivitet av typen HDInsightHive i avsnittet activities. I det här exemplet transformerar HDInsight Hive-aktiviteten data från Azure Blob Storage genom att köra en Hive-skriptfil på ett Azure HDInsight Hadoop-kluster.

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

Observera följande punkter:

• I activities-avsnittet finns det bara en aktivitet vars typ anges till HDInsightHive.
• Hive-skriptfilen partitionweblogs.hql lagras i Azure Storage-kontot (anges av scriptLinkedService, kallas AzureStorageLinkedService) och i skriptmappen i containern adfgetstarted .
• Avsnittet defines används för att ange körningsinställningar som skickas till Hive-skriptet som Hive-konfigurationsvärden (till exempel ${hiveconf:inputtable}, ${hiveconf:partitionedtable}.

Avsnittet typeProperties är olika för varje transformeringsaktivitet. Om du vill ha mer information om vilka typegenskaper som stöds för en transformeringsaktivitet klickar du på transformeringsaktiviteten i Datatransformeringsaktiviteter.

En fullständig genomgång av hur du skapar denna pipeline finns i Självstudier: transformera data med Spark.

Flera aktiviteter i en pipeline

De två föregående exemplen innehåller bara en aktivitet. Du kan fler än en aktivitet i en pipeline. Om du har flera aktiviteter i en pipeline och efterföljande aktiviteter inte är beroende av tidigare aktiviteter kan aktiviteterna köras parallellt.

Du kan länka två aktiviteter genom att använda aktivitetsberoende, som definierar hur efterföljande aktiviteter är beroende av tidigare aktiviteter, vilket fastställer villkoret för om nästa uppgift ska köras. En aktivitet kan vara beroende av en eller flera tidigare aktiviteter med olika beroendevillkor.

Schemaläggningspipelines

Pipelines schemaläggs av utlösare. Det finns olika typer av utlösare (Scheduler-utlösare, som gör att pipelines kan utlösas enligt ett klocktidsschema (wall-clock) samt den manuella utlösaren, som utlöser pipelines på begäran). Mer information om utlösare finns i artikeln om pipelinekörning och utlösare.

Om du vill att utlösaren startar en pipelinekörning måste du ta med en pipelinereferens i utlösardefinitionen. Pipelines och utlösare har ett n-m-förhållande. Flera utlösare kan starta en enda pipeline, och samma utlösare kan starta flera pipelines. När utlösaren har definierats måste du starta utlösaren så att den kan börja utlösa pipeline. Mer information om utlösare finns i artikeln om pipelinekörning och utlösare.

Säg till exempel att du har en Scheduler-utlösare, "Trigger A", som jag vill starta min pipeline, "MyCopyPipeline". Du definierar utlösaren, som du ser i följande exempel:

Definition av TriggerA

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

Nästa steg

I följande självstudier får du stegvisa instruktioner för att skapa pipelines med aktiviteter:

• Skapa en pipeline med en kopieringsaktivitet
• Skapa en pipeline med en datatransformeringsaktivitet

Så här uppnår du CI/CD (kontinuerlig integrering och leverans) med hjälp av Azure Data Factory

• Kontinuerlig integrering och leverans i Azure Data Factory