DevOps i en pipeline för datainmatning

  • Artikel
  • 10 minuter för att läsa

I de flesta fall är en datainmatningslösning en sammansättning av skript, tjänstanrop och en pipeline som orkestrering av alla aktiviteter. I den här artikeln får du lära dig hur du använder DevOps-metoder för utvecklingslivscykeln för en gemensam pipeline för datainmatning som förbereder data för träning av maskininlärningsmodeller. Pipelinen bygger på följande Azure-tjänster:

  • Azure Data Factory: Läser rådata och orkestrering av data.
  • Azure Databricks: Kör en Python-notebook-dator som transformerar data.
  • Azure Pipelines: Automatiserar en kontinuerlig integrerings- och utvecklingsprocess.

Arbetsflöde för datainmatningspipeline

Pipelinen för datainmatning implementerar följande arbetsflöde:

  1. Rådata läses in i en Azure Data Factory-pipeline (ADF).
  2. ADF-pipelinen skickar data till ett Azure Databricks som kör en Python-notebook-dator för att transformera data.
  3. Data lagras i en blobcontainer, där de kan användas av Azure Machine Learning för att träna en modell.

arbetsflöde för datainmatningspipeline

Översikt över kontinuerlig integrering och leverans

Som med många programvarulösningar finns det ett team (till exempel datatekniker) som arbetar med den. De samarbetar och delar samma Azure-resurser som Azure Data Factory, Azure Databricks och Azure Storage konton. Samlingen av dessa resurser är en utvecklingsmiljö. Datateknikern bidrar till samma källkodsbas.

Ett system för kontinuerlig integrering och leverans automatiserar processen med att skapa, testa och leverera (distribuera) lösningen. Processen för kontinuerlig integrering (CI) utför följande uppgifter:

  • Sätter ihop koden
  • Kontrollerar den med kodkvalitetstester
  • Kör enhetstester
  • Skapar artefakter som testad kod och Azure Resource Manager mallar

Processen kontinuerlig leverans (CD) distribuerar artefakterna till nedströmsmiljöerna.

Diagram över inmatning av cicd-data

Den här artikeln visar hur du automatiserar CI- och CD-processerna med Azure Pipelines.

Hantering av källkontroll

Källkontrollhantering krävs för att spåra ändringar och möjliggöra samarbete mellan teammedlemmar. Koden lagras till exempel på en Azure DevOps-, GitHub- eller GitLab-lagringsplats. Arbetsflödet för samarbete baseras på en förgreningsmodell. Till exempel GitFlow.

Python Notebook-källkod

Datateknikern arbetar med Python Notebook-källkoden antingen lokalt i en IDE (till exempel Visual Studio Code)eller direkt på Databricks-arbetsytan. När kodändringarna är klara sammanfogas de till lagringsplatsen enligt en förgreningsprincip.

Tips

Vi rekommenderar att du lagrar koden i .py filer i stället för i Jupyter Notebook .ipynb format. Det förbättrar kodens läsbarhet och möjliggör automatiska kodkvalitetskontroller i CI-processen.

Azure Data Factory källkod

Källkoden för Azure Data Factory är en samling JSON-filer som genereras av en Azure Data Factory arbetsyta. Normalt arbetar datatekniker med en visuell designer på Azure Data Factory i stället för direkt med källkodsfilerna.

Information om hur du konfigurerar arbetsytan för att använda en lagringsplats för källkontroll finns i Author with Azure Repos Git integration (Skapa med Azure Repos Git-integrering).

Kontinuerlig integrering (CI)

Slutmålet med den kontinuerliga integreringsprocessen är att samla in det gemensamma teamets arbete från källkoden och förbereda det för distributionen till nedströmsmiljöerna. Precis som med källkodshanteringen är den här processen annorlunda för Python-anteckningsböcker och Azure Data Factory pipelines.

Python Notebook CI

CI-processen för Python Notebooks hämtar koden från samarbetsgrenen (till exempel master _ eller _utveckla**) och utför följande aktiviteter:

  • Kodlintning
  • Enhetstestning
  • Spara koden som en artefakt

Följande kodfragment visar implementeringen av de här stegen i en Azure DevOps yaml-pipeline:

steps:
- script: |
   flake8 --output-file=$(Build.BinariesDirectory)/lint-testresults.xml --format junit-xml  
  workingDirectory: '$(Build.SourcesDirectory)'
  displayName: 'Run flake8 (code style analysis)'  
  
- script: |
   python -m pytest --junitxml=$(Build.BinariesDirectory)/unit-testresults.xml $(Build.SourcesDirectory)
  displayName: 'Run unit tests'

- task: PublishTestResults@2
  condition: succeededOrFailed()
  inputs:
    testResultsFiles: '$(Build.BinariesDirectory)/*-testresults.xml'
    testRunTitle: 'Linting & Unit tests'
    failTaskOnFailedTests: true
  displayName: 'Publish linting and unit test results'

- publish: $(Build.SourcesDirectory)
    artifact: di-notebooks

Pipelinen använder flake8 för att göra Python-kod lintningen. Den kör enhetstesterna som definierats i källkoden och publicerar linting- och testresultaten så att de är tillgängliga på Azure Pipeline-körningsskärmen:

lintenhetstester

Om lint- och enhetstestningen lyckas kopierar pipelinen källkoden till artefaktdatabasen som ska användas av efterföljande distributionssteg.

Azure Data Factory CI

CI-processen för en Azure Data Factory-pipeline är en flaskhals för en pipeline för datainmatning. Det finns ingen kontinuerlig integrering. En distribuerad artefakt för Azure Data Factory är en samling Azure Resource Manager mallar. Det enda sättet att skapa dessa mallar är att klicka på publiceringsknappen i Azure Data Factory arbetsytan.

  1. Datatekniker sammanfogar källkoden från sina funktionsgrenar i samarbetsgrenen, till exempel master _ eller _utveckla**.
  2. Någon med de beviljade behörigheterna klickar på publiceringsknappen för Azure Resource Manager skapa mallar från källkoden i samarbetsgrenen.
  3. Arbetsytan verifierar pipelines (se det som linting- och enhetstestning), genererar Azure Resource Manager-mallar (tänk på det som en byggnad) och sparar de genererade mallarna till en teknisk gren adf_publish i samma kodlager (tänk på det när du publicerar artefakter). Den här grenen skapas automatiskt av Azure Data Factory arbetsytan.

Mer information om den här processen finns i Kontinuerlig integrering och leverans i Azure Data Factory.

Det är viktigt att se till att de Azure Resource Manager mallarna är miljöoberoende. Det innebär att alla värden som kan skilja sig åt mellan miljöer parametriseras. Azure Data Factory är tillräckligt smart för att exponera de flesta sådana värden som parametrar. I följande mall visas till exempel anslutningsegenskaperna för en Azure Machine Learning arbetsyta som parametrar:

{
    "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentParameters.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "factoryName": {
            "value": "devops-ds-adf"
        },
        "AzureMLService_servicePrincipalKey": {
            "value": ""
        },
        "AzureMLService_properties_typeProperties_subscriptionId": {
            "value": "0fe1c235-5cfa-4152-17d7-5dff45a8d4ba"
        },
        "AzureMLService_properties_typeProperties_resourceGroupName": {
            "value": "devops-ds-rg"
        },
        "AzureMLService_properties_typeProperties_servicePrincipalId": {
            "value": "6e35e589-3b22-4edb-89d0-2ab7fc08d488"
        },
        "AzureMLService_properties_typeProperties_tenant": {
            "value": "72f988bf-86f1-41af-912b-2d7cd611db47"
        }
    }
}

Men du kanske vill exponera dina anpassade egenskaper som inte hanteras av Azure Data Factory som standard. I scenariot med den här artikeln anropar Azure Data Factory en Python-notebook-dator som bearbetar data. Notebook-filen accepterar en parameter med namnet på en indatafil.

import pandas as pd
import numpy as np

data_file_name = getArgument("data_file_name")
data = pd.read_csv(data_file_name)

labels = np.array(data['target'])
...

Det här namnet skiljer sig för Dev _, _QA,_ _UAT_ och _PROD_ miljöer. I en komplex pipeline med flera aktiviteter kan det finnas flera anpassade egenskaper. Det är bra att samla in alla dessa värden på ett ställe och definiera dem som pipeline _variabler**:

Skärmbild som visar en notebook-fil med namnet PrepareData och M L Execute Pipeline med namnet M L Execute Pipeline längst upp med fliken Variabler markerad nedan med alternativet att lägga till nya variabler, var och en med ett namn, en typ och ett standardvärde.

Pipeline-aktiviteterna kan referera till pipelinevariablerna när de faktiskt används:

Skärmbild som visar en notebook-dator med namnet PrepareData och M L Execute Pipeline med namnet M L Execute Pipeline längst upp Inställningar fliken nedan.

Arbetsytan Azure Data Factory exponerar inte pipelinevariabler som Azure Resource Manager som standard. Arbetsytan använder standardparameteriseringsmallen som anger vilka pipelineegenskaper som ska exponeras som Azure Resource Manager mallparametrar. Om du vill lägga till pipelinevariabler i listan uppdaterar du avsnittet i mallen för standardparameterisering med följande kodfragment och placerar json-resultatfilen i roten för "Microsoft.DataFactory/factories/pipelines" källmappen: 

"Microsoft.DataFactory/factories/pipelines": {
        "properties": {
            "variables": {
                "*": {
                    "defaultValue": "="
                }
            }
        }
    }

Om du gör det Azure Data Factory arbetsytan att lägga till variablerna i parameterlistan när du klickar på publiceringsknappen:

{
    "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentParameters.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "factoryName": {
            "value": "devops-ds-adf"
        },
        ...
        "data-ingestion-pipeline_properties_variables_data_file_name_defaultValue": {
            "value": "driver_prediction_train.csv"
        }        
    }
}

Värdena i JSON-filen är standardvärden som konfigurerats i pipelinedefinitionen. De förväntas åsidosättas med målmiljövärdena när Azure Resource Manager distribueras.

Kontinuerlig leverans (CD)

Kontinuerlig leverans tar artefakterna och distribuerar dem till den första målmiljön. Det ser till att lösningen fungerar genom att köra tester. Om det lyckas fortsätter den till nästa miljö.

CD Azure Pipeline består av flera steg som representerar miljöerna. Varje steg innehåller distributioner och jobb som utför följande steg:

  • Distribuera en Python Notebook till Azure Databricks arbetsyta
  • Distribuera en Azure Data Factory pipeline
  • Köra en pipeline
  • Kontrollera datainmatningsresultatet

Pipelinestegen kan konfigureras med godkännanden och portar som ger ytterligare kontroll över hur distributionsprocessen utvecklas genom kedjan av miljöer.

Distribuera en Python Notebook

Följande kodfragment definierar en Azure Pipeline-distribution som kopierar en Python-notebook-fil till ett Databricks-kluster:

- stage: 'Deploy_to_QA'
  displayName: 'Deploy to QA'
  variables:
  - group: devops-ds-qa-vg
  jobs:
  - deployment: "Deploy_to_Databricks"
    displayName: 'Deploy to Databricks'
    timeoutInMinutes: 0
    environment: qa
    strategy:
      runOnce:
        deploy:
          steps:
            - task: UsePythonVersion@0
              inputs:
                versionSpec: '3.x'
                addToPath: true
                architecture: 'x64'
              displayName: 'Use Python3'

            - task: configuredatabricks@0
              inputs:
                url: '$(DATABRICKS_URL)'
                token: '$(DATABRICKS_TOKEN)'
              displayName: 'Configure Databricks CLI'    

            - task: deploynotebooks@0
              inputs:
                notebooksFolderPath: '$(Pipeline.Workspace)/di-notebooks'
                workspaceFolder: '/Shared/devops-ds'
              displayName: 'Deploy (copy) data processing notebook to the Databricks cluster'

Artefakterna som produceras av KI kopieras automatiskt till distributionsagenten och är tillgängliga i $(Pipeline.Workspace) mappen . I det här fallet refererar distributionsuppgiften till di-notebooks artefakten som innehåller Python-anteckningsboken. Den här distributionen använder Databricks Azure DevOps-tillägget för att kopiera notebook-filerna till Databricks-arbetsytan.

Fasen Deploy_to_QA innehåller en referens till variabelgruppen som devops-ds-qa-vg definierats i Azure DevOps-projektet. Stegen i det här steget refererar till variablerna från den här variabelgruppen (till exempel $(DATABRICKS_URL) och $(DATABRICKS_TOKEN) ). Tanken är att nästa steg (till exempel ) ska fungera med samma variabelnamn som definierats i en egen Deploy_to_UAT UAT-begränsad variabelgrupp.

Distribuera en Azure Data Factory pipeline

En distribuerad artefakt för Azure Data Factory är en Azure Resource Manager mall. Den kommer att distribueras med azure-resursgruppens distributionsuppgift, vilket visas i följande kodfragment:

  - deployment: "Deploy_to_ADF"
    displayName: 'Deploy to ADF'
    timeoutInMinutes: 0
    environment: qa
    strategy:
      runOnce:
        deploy:
          steps:
            - task: AzureResourceGroupDeployment@2
              displayName: 'Deploy ADF resources'
              inputs:
                azureSubscription: $(AZURE_RM_CONNECTION)
                resourceGroupName: $(RESOURCE_GROUP)
                location: $(LOCATION)
                csmFile: '$(Pipeline.Workspace)/adf-pipelines/ARMTemplateForFactory.json'
                csmParametersFile: '$(Pipeline.Workspace)/adf-pipelines/ARMTemplateParametersForFactory.json'
                overrideParameters: -data-ingestion-pipeline_properties_variables_data_file_name_defaultValue "$(DATA_FILE_NAME)"

Värdet för parametern datafilnamn kommer från variabeln som $(DATA_FILE_NAME) definierats i en QA-fasvariabelgrupp. På samma sätt kan alla parametrar som definieras i ARMTemplateForFactory.json åsidosättas. Om de inte är det används standardvärdena.

Kör pipelinen och kontrollera datainmatningsresultatet

Nästa steg är att se till att den distribuerade lösningen fungerar. Följande jobbdefinition kör en Azure Data Factory pipeline med ett PowerShell-skript och kör en Python-notebook-Azure Databricks ett kluster. Notebook-filen kontrollerar om data har matats in korrekt och validerar resultatdatafilen med $(bin_FILE_NAME) namn.

  - job: "Integration_test_job"
    displayName: "Integration test job"
    dependsOn: [Deploy_to_Databricks, Deploy_to_ADF]
    pool:
      vmImage: 'ubuntu-latest'
    timeoutInMinutes: 0
    steps:
    - task: AzurePowerShell@4
      displayName: 'Execute ADF Pipeline'
      inputs:
        azureSubscription: $(AZURE_RM_CONNECTION)
        ScriptPath: '$(Build.SourcesDirectory)/adf/utils/Invoke-ADFPipeline.ps1'
        ScriptArguments: '-ResourceGroupName $(RESOURCE_GROUP) -DataFactoryName $(DATA_FACTORY_NAME) -PipelineName $(PIPELINE_NAME)'
        azurePowerShellVersion: LatestVersion
    - task: UsePythonVersion@0
      inputs:
        versionSpec: '3.x'
        addToPath: true
        architecture: 'x64'
      displayName: 'Use Python3'

    - task: configuredatabricks@0
      inputs:
        url: '$(DATABRICKS_URL)'
        token: '$(DATABRICKS_TOKEN)'
      displayName: 'Configure Databricks CLI'    

    - task: executenotebook@0
      inputs:
        notebookPath: '/Shared/devops-ds/test-data-ingestion'
        existingClusterId: '$(DATABRICKS_CLUSTER_ID)'
        executionParams: '{"bin_file_name":"$(bin_FILE_NAME)"}'
      displayName: 'Test data ingestion'

    - task: waitexecution@0
      displayName: 'Wait until the testing is done'

Den sista uppgiften i jobbet kontrollerar resultatet av notebook-körningen. Om den returnerar ett fel anges statusen för pipelinekörningen till misslyckad.

Sätta ihop bitar

Den fullständiga CI/CD Azure Pipeline består av följande steg:

  • CI
  • Distribuera till QA
    • Distribuera till Databricks + Distribuera till ADF
    • Integrationstest

Den innehåller ett antal Distribuera _ faser som motsvarar antalet målmiljöer som du har. Varje _ Distribuera steg innehåller två distributioner som körs parallellt och ett jobb som körs efter distributioner för att testa lösningen i miljön.

En exempelimplementering av pipelinen sammanställs i följande yaml-kodfragment:

variables:
- group: devops-ds-vg

stages:
- stage: 'CI'
  displayName: 'CI'
  jobs:
  - job: "CI_Job"
    displayName: "CI Job"
    pool:
      vmImage: 'ubuntu-latest'
    timeoutInMinutes: 0
    steps:
    - task: UsePythonVersion@0
      inputs:
        versionSpec: '3.x'
        addToPath: true
        architecture: 'x64'
      displayName: 'Use Python3'
    - script: pip install --upgrade flake8 flake8_formatter_junit_xml
      displayName: 'Install flake8'
    - checkout: self
    - script: |
       flake8 --output-file=$(Build.BinariesDirectory)/lint-testresults.xml --format junit-xml  
    workingDirectory: '$(Build.SourcesDirectory)'
    displayName: 'Run flake8 (code style analysis)'  
    - script: |
       python -m pytest --junitxml=$(Build.BinariesDirectory)/unit-testresults.xml $(Build.SourcesDirectory)
    displayName: 'Run unit tests'
    - task: PublishTestResults@2
    condition: succeededOrFailed()
    inputs:
        testResultsFiles: '$(Build.BinariesDirectory)/*-testresults.xml'
        testRunTitle: 'Linting & Unit tests'
        failTaskOnFailedTests: true
    displayName: 'Publish linting and unit test results'    

    # The CI stage produces two artifacts (notebooks and ADF pipelines).
    # The pipelines Azure Resource Manager templates are stored in a technical branch "adf_publish"
    - publish: $(Build.SourcesDirectory)/$(Build.Repository.Name)/code/dataingestion
      artifact: di-notebooks
    - checkout: git://${{variables['System.TeamProject']}}@adf_publish    
    - publish: $(Build.SourcesDirectory)/$(Build.Repository.Name)/devops-ds-adf
      artifact: adf-pipelines

- stage: 'Deploy_to_QA'
  displayName: 'Deploy to QA'
  variables:
  - group: devops-ds-qa-vg
  jobs:
  - deployment: "Deploy_to_Databricks"
    displayName: 'Deploy to Databricks'
    timeoutInMinutes: 0
    environment: qa
    strategy:
      runOnce:
        deploy:
          steps:
            - task: UsePythonVersion@0
              inputs:
                versionSpec: '3.x'
                addToPath: true
                architecture: 'x64'
              displayName: 'Use Python3'

            - task: configuredatabricks@0
              inputs:
                url: '$(DATABRICKS_URL)'
                token: '$(DATABRICKS_TOKEN)'
              displayName: 'Configure Databricks CLI'    

            - task: deploynotebooks@0
              inputs:
                notebooksFolderPath: '$(Pipeline.Workspace)/di-notebooks'
                workspaceFolder: '/Shared/devops-ds'
              displayName: 'Deploy (copy) data processing notebook to the Databricks cluster'             
  - deployment: "Deploy_to_ADF"
    displayName: 'Deploy to ADF'
    timeoutInMinutes: 0
    environment: qa
    strategy:
      runOnce:
        deploy:
          steps:
            - task: AzureResourceGroupDeployment@2
              displayName: 'Deploy ADF resources'
              inputs:
                azureSubscription: $(AZURE_RM_CONNECTION)
                resourceGroupName: $(RESOURCE_GROUP)
                location: $(LOCATION)
                csmFile: '$(Pipeline.Workspace)/adf-pipelines/ARMTemplateForFactory.json'
                csmParametersFile: '$(Pipeline.Workspace)/adf-pipelines/ARMTemplateParametersForFactory.json'
                overrideParameters: -data-ingestion-pipeline_properties_variables_data_file_name_defaultValue "$(DATA_FILE_NAME)"
  - job: "Integration_test_job"
    displayName: "Integration test job"
    dependsOn: [Deploy_to_Databricks, Deploy_to_ADF]
    pool:
      vmImage: 'ubuntu-latest'
    timeoutInMinutes: 0
    steps:
    - task: AzurePowerShell@4
      displayName: 'Execute ADF Pipeline'
      inputs:
        azureSubscription: $(AZURE_RM_CONNECTION)
        ScriptPath: '$(Build.SourcesDirectory)/adf/utils/Invoke-ADFPipeline.ps1'
        ScriptArguments: '-ResourceGroupName $(RESOURCE_GROUP) -DataFactoryName $(DATA_FACTORY_NAME) -PipelineName $(PIPELINE_NAME)'
        azurePowerShellVersion: LatestVersion
    - task: UsePythonVersion@0
      inputs:
        versionSpec: '3.x'
        addToPath: true
        architecture: 'x64'
      displayName: 'Use Python3'

    - task: configuredatabricks@0
      inputs:
        url: '$(DATABRICKS_URL)'
        token: '$(DATABRICKS_TOKEN)'
      displayName: 'Configure Databricks CLI'    

    - task: executenotebook@0
      inputs:
        notebookPath: '/Shared/devops-ds/test-data-ingestion'
        existingClusterId: '$(DATABRICKS_CLUSTER_ID)'
        executionParams: '{"bin_file_name":"$(bin_FILE_NAME)"}'
      displayName: 'Test data ingestion'

    - task: waitexecution@0
      displayName: 'Wait until the testing is done'

Nästa steg