Pipelines ve çalışma Azure Data Factory etkinlikler Azure Synapse Analytics

Uygulama hedefi: Azure SYNAPSE Analytics 'i Azure Data Factory

Bu makale, Azure Data Factory ve Azure Synapse Analytics'daki işlem hatlarını ve etkinlikleri anlamanıza yardımcı olur ve bunları veri taşıma ve veri işleme senaryolarınız için uz- uç veri odaklı iş akışları oluşturmak için kullanır.

Genel Bakış

Bir Data Factory veya Synapse Çalışma Alanı'nın bir veya daha fazla işlem hattı olabilir. İşlem hattı, bir görevi gerçekleştiren etkinliklerden oluşan mantıksal gruptur. Örneğin bir işlem hattında günlük verilerini alıp temizleyen ve ardından bu verilerin analiz edilmesi için bir eşleme veri akışı başlatan etkinlikler bulunabilir. İşlem hattı, etkinlikleri ayrı ayrı değil küme halinde yönetmenizi sağlar. Etkinlikleri tek tek zamanlamak yerine işlem hattını dağıtabilir ve zamanlayabilirsiniz.

İşlem hattındaki etkinlikler, verilerinizde gerçekleştirilecek eylemleri tanımlar. Örneğin, verileri bir Azure Blob depolama SQL Server kopyalamak için kopyalama etkinliği Depolama. Ardından bir veri akışı etkinliği veya Databricks Not Defteri etkinliği kullanarak verileri blob depolamadan iş zekası raporlama çözümlerinin der Azure Synapse Analytics bir havuza iş akışından dönüştürebilirsiniz.

Azure Data Factory Azure Synapse Analytics üç etkinlik grubu vardır: veri taşıma etkinlikleri,veri dönüştürme etkinliklerive denetim etkinlikleri. Bir etkinlik sıfır veya daha fazla giriş veri kümesi alıp bir veya daha fazla çıkış veri kümesi üretebilir. Aşağıdaki diyagramda işlem hattı, etkinlik ve veri kümesi arasındaki ilişki gösterir:

Giriş veri kümesi işlem hattında bir etkinliğin girişini, çıktı veri kümesi ise etkinliğin çıktısını temsil eder. Veri kümeleri tablolar, dosyalar, klasörler ve belgeler gibi farklı veri depolarındaki verileri tanımlar. Bir veri kümesi oluşturduktan sonra, bu kümeyi bir işlem hattındaki etkinliklerle birlikte kullanabilirsiniz. Örneğin, veri kümesi bir Kopyalama Etkinliğinin veya HDInsightHive Etkinliğinin giriş/çıkış veri kümesi olabilir. Veri kümeleri hakkında daha fazla bilgi için Azure Data Factory'de Veri Kümeleri makalesine bakın.

Veri taşıma etkinlikleri

Data Factory’deki Kopyalama Etkinliği bir kaynak veri deposundan havuz veri deposuna verileri kopyalar. Data Factory bu bölümdeki tabloda listelenen veri depolarını destekler. Herhangi bir kaynaktan gelen veriler herhangi bir havuza yazılabilir. Bir depoya veya depodan veri kopyalama hakkında bilgi edinmek için veri deposuna tıklayın.

Daha fazla bilgi için Kopyalama Etkinliği - Genel Bakış makalesine bakın.

Veri dönüştürme etkinlikleri

Azure Data Factory ve Azure Synapse Analytics tek eklensin veya başka bir etkinlikle zincirleme aşağıdaki dönüştürme etkinliklerini destekler.

Daha fazla bilgi için veri dönüştürme etkinlikleri makalesine bakın.

Denetim akışı etkinlikleri

Aşağıdaki denetim akışı etkinlikleri desteklenir:

İşlem Hattı JSON

JSON biçiminde işlem hattı şöyle tanımlanır:

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

Etkinlik JSON

Etkinlikler bölümünde tanımlanmış bir veya daha fazla etkinlik olabilir. İki temel etkinlik türü vardır: Yürütme ve Denetim Etkinlikleri.

Yürütme etkinlikleri

Yürütme etkinlikleri veri taşıma ve veri dönüştürme etkinliklerini içerir. Aşağıdaki üst düzey yapıya sahiptir:

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

Aşağıdaki tabloda, etkinlik JSON tanımındaki özellikler açıklamaktadır:

Etkinlik ilkesi

İlkeler bir etkinliğin çalışma zamanı davranışını etkiler ve yapılandırma seçenekleri sunar. Etkinlik İlkeleri yalnızca yürütme etkinlikleri için kullanılabilir.

Etkinlik ilkesi JSON tanımı

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

Denetim etkinliği

Denetim etkinlikleri aşağıdaki üst düzey yapıya sahiptir:

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

Etkinlik bağımlılığı

Etkinlik Bağımlılığı, sonraki etkinliklerin önceki etkinliklere nasıl bağımlı olduğunu tanımlar ve sonraki görevi yürütmeye devam edip olmadığı koşullarını belirler. Bir etkinlik farklı bağımlılık koşullarıyla daha önceki bir veya birden çok etkinliğe bağımlı olabilir.

Farklı bağımlılık koşulları şunlardır: Başarılı, Başarısız, Atlandı, Tamamlandı.

Örneğin, bir işlem hattında Etkinlik A -> Etkinlik B ise oluşabilecek farklı senaryolar şunlardır:

• Etkinlik B, Etkinlik A’ya başarılı koşuluyla bağımlıdır: Etkinlik B yalnızca Etkinlik A’nın son durumu başarılı ise çalışır
• Etkinlik B, Etkinlik A’ya başarısız koşuluyla bağımlıdır: Etkinlik B yalnızca Etkinlik A’nın son durumu başarısız ise çalışır
• Etkinlik B, Etkinlik A’ya tamamlandı koşuluyla bağımlıdır: Etkinlik B yalnızca Etkinlik A’nın son durumu başarılı veya başarısız ise çalışır
• Etkinlik B, Etkinlik A'da atlanan bir bağımlılık koşuluna sahip: Etkinlik B, Etkinlik A'nın son durumu atlandı ise çalışır. Atlandı koşulu, her bir etkinliğin yalnızca önceki etkinlik başarılı olursa çalıştığı Etkinlik X -> Etkinlik Y -> Etkinlik Z senaryosunda gerçekleşir. Etkinlik X başarısız olursa, Etkinlik Y hiçbir zaman yürütülmezse "Atlandı" durumuna sahip olur. Benzer şekilde, Z Etkinliği de "Atlandı" durumuna sahip.

Örnek: Etkinlik 2, Etkinlik 1’in başarılı olmasına bağlıdır

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

Örnek kopyalama işlem hattı

Aşağıdaki örnek işlem hattında, Etkinlikler bölümünde Kopyalama türünde olan bir etkinlik vardır. Bu örnekte kopyalama etkinliği, verileri Azure Blob depolamadan azure blob depolama Azure SQL Veritabanı.

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

Aşağıdaki noktalara dikkat edin:

• Etkinlikler bölümünde, türü****Copy olarak ayarlanmış yalnızca bir etkinlik vardır.
• Etkinlik girdisi InputDataset olarak, etkinlik çıktısı ise OutputDataset olarak ayarlanmıştır. JSON biçiminde veri kümeleri tanımlamak için Veri Kümeleri makalesine bakın.
• typeProperties bölümünde BlobSource kaynak türü, SqlSink de havuz türü olarak belirtilir. Veri taşıma etkinlikleri bölümünde, verileri veri deposuna/veri deposundan taşıma hakkında daha fazla bilgi almak için kaynak veya havuz olarak kullanmak istediğiniz veri deposuna tıklayın.

Bu işlem hattını oluşturma işleminin eksiksiz bir yolu için bkz. Hızlı Başlangıç: Data Factory.

Örnek dönüştürme işlem hattı

Aşağıdaki örnek işlem hattında, etkinlikler bölümünde HDInsightHive türünde olan bir etkinlik vardır. Bu örnekte HDInsight Hive etkinliği, bir Azure HDInsight Hadoop kümesinde Hive betik dosyası çalıştırarak verileri bir Azure Blob depolamadan dönüştürür.

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

Aşağıdaki noktalara dikkat edin:

• Etkinlikler bölümünde türü****HDInsightHive olarak ayarlanmış yalnızca bir etkinlik vardır.
• partitionweblogs.hql Hive betik dosyası Azure Depolama hesabında (scriptLinkedService tarafından belirtilen AzureStorageLinkedService olarak adlandırılan) ve kapsayıcının betik klasöründe adfgetstarted depolanır.
• defines bölümü, hive betiğine Hive yapılandırma değerleri olarak (örn {hiveconf:inputtable}, ${hiveconf:partitionedtable}) geçirilen çalışma zamanı ayarlarını belirtmek için kullanılır.

TypeProperties bölümü her bir dönüştürme etkinliği için farklıdır. Bir dönüştürme etkinliği için desteklenen tür özellikleri hakkında bilgi edinmek için Veri dönüştürme etkinlikleri içindeki dönüştürme etkinliklerine tıklayın.

Bu işlem hattını oluşturmak üzere izlenecek tam yol için bkz. Öğretici: Spark kullanarak veri dönüştürme.

Bir işlem hattında birden çok etkinlik

Önceki iki örnekte işlem hatları yalnızca bir etkinlik içeriyordu. Bir işlem hattında birden fazla etkinliğiniz olabilir. Bir işlem hattında birden fazla etkinlik varsa ve sonraki etkinlikler önceki etkinliklere bağımlı değilse, etkinlikler paralel olarak çalışabilir.

Sonraki etkinliklerin önceki etkinliklere nasıl bağımlı olduğunu tanımlayan ve sonraki görevi yürütmeye devam edilip edilmeyeceğine yönelik koşulu belirleyen etkinlik bağımlılığını kullanarak iki etkinliği zincirleyebilirsiniz. Bir etkinlik farklı bağımlılık koşullarıyla daha önceki bir veya daha çok etkinliğe bağımlı olabilir.

İşlem hatlarını zamanlama

İşlem hatları tetikleyiciler tarafından zamanlanır. Farklı türlerde tetikleyiciler vardır (İşlem hatlarının duvar saati zamanlaması ile tetiklenen zamanlayıcı tetikleyicisi ve işlem hatlarını isteğe bağlı olarak tetikleyen el ile tetikleyici). Tetikleyiciler hakkında daha fazla bilgi için işlem hattı yürütme ve tetikleyicileri makalesine bakın.

Tetikleyicinizin bir işlem hattı çalıştırmasını başlatması için tetikleyici tanımındaki belirli işlem hattının işlem hattı başvurusunu eklemeniz gerekir. İşlem hatları ve tetikleyiciler n-m ilişkisine sahiptir. Birden çok tetikleyici tek bir işlem hattını, bir tetikleyici de birden fazla işlem hattını başlatabilir. Tetikleyici tanımlandıktan sonra işlem hattını tetiklemesini başlatmak için tetikleyiciyi başlatmanız gerekir. Tetikleyiciler hakkında daha fazla bilgi için işlem hattı yürütme ve tetikleyicileri makalesine bakın.

Örneğin, "MyCopyPipeline" işlem hattımı tetiklemek istediğiniz "Tetikleyici A" zamanlayıcı tetikleyicisi olduğunu diyelim. Tetikleyiciyi aşağıdaki örnekte gösterildiği gibi tanımlarsiniz:

Tetikleyici A tanımı

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

Sonraki adımlar

Etkinliklerle işlem hatları oluşturmaya yönelik adım adım yönergeler için aşağıdaki öğreticilere bakın:

• Kopyalama etkinliği içeren işlem hattı oluşturma
• Veri dönüştürme etkinliğine sahip işlem hattı oluşturma

Azure Data Factory kullanarak CI/CD (sürekli tümleştirme ve teslim) Azure Data Factory

• Azure Data Factory’de sürekli tümleştirme ve teslim