Gegevens Flow activiteit in Azure Data Factory en Azure Synapse Analytics

VAN TOEPASSING OP: Azure Data Factory Azure Synapse Analytics

Gebruik de Data Flow om gegevens te transformeren en te verplaatsen via toewijzingsgegevensstromen. Als u geen tijd hebt voor gegevensstromen, zie Toewijzingsgegevens Flow overzicht

Syntax

{
    "name": "MyDataFlowActivity",
    "type": "ExecuteDataFlow",
    "typeProperties": {
      "dataflow": {
         "referenceName": "MyDataFlow",
         "type": "DataFlowReference"
      },
      "compute": {
         "coreCount": 8,
         "computeType": "General"
      },
      "traceLevel": "Fine",
      "runConcurrently": true,
      "continueOnError": true,      
      "staging": {
          "linkedService": {
              "referenceName": "MyStagingLinkedService",
              "type": "LinkedServiceReference"
          },
          "folderPath": "my-container/my-folder"
      },
      "integrationRuntime": {
          "referenceName": "MyDataFlowIntegrationRuntime",
          "type": "IntegrationRuntimeReference"
      }
}

Typeeigenschappen

Dynamisch gegevensstroomrekengrootte tijdens runtime

De eigenschappen Core Count en Compute Type kunnen dynamisch worden ingesteld om aan te passen aan de grootte van uw binnenkomende brongegevens tijdens runtime. Gebruik pijplijnactiviteiten zoals Opzoek of Metagegevens opschonen om de grootte van de gegevens van de brongegevensset te vinden. Gebruik vervolgens Dynamische inhoud toevoegen in de activiteitseigenschappen Flow Data Flow.

Hier is een korte videozelfstudie waarin deze techniek wordt uitgelegd

Data Flow Integration Runtime

Kies welke Integration Runtime moet worden gebruikt voor de uitvoering van uw Flow activiteit. De service gebruikt standaard de Azure Integration-runtime voor automatisch oplossen met vier werkkernen. Deze IR heeft een rekentype voor algemeen gebruik en wordt uitgevoerd in dezelfde regio als uw service-exemplaar. Voor operationele pijplijnen is het raadzaam om uw eigen Azure Integration Runtimes te maken die specifieke regio's, rekentype, kernen en TTL definiëren voor het uitvoeren van uw gegevensstroomactiviteit.

Een minimaal rekentype van Algemeen (geoptimaliseerd voor rekenkracht wordt niet aanbevolen voor grote workloads) met een configuratie van 8+8 (16 in totaal v-cores) en een 10 minuten is de minimale aanbeveling voor de meeste productieworkloads. Door een kleine TTL in te stellen, kan Azure IR een warm cluster onderhouden dat niet de enkele minuten van de starttijd voor een niet-koude cluster in stand houdt. U kunt de uitvoering van uw gegevensstromen nog sneller uitvoeren door 'Snel opnieuw gebruiken' te selecteren in de configuraties Azure IR gegevensstroom. Zie Azure Integration Runtime voor meer informatie.

PolyBase

Als u een testlocatie gebruikt Azure Synapse Analytics sink of bron, moet u een faseringslocatie kiezen voor uw PolyBase-batchbelasting. PolyBase maakt het bulksgewijs laden van gegevens mogelijk in plaats van de gegevens rij voor rij te laden. PolyBase vermindert de laadtijd in de Azure Synapse Analytics.

Niveau van logboekregistratie

Als u niet elke pijplijnuitvoering van uw gegevensstroomactiviteiten nodig hebt om alle uitgebreide telemetrielogboeken volledig te registreren, kunt u eventueel uw logboekregistratieniveau instellen op Basic of Geen. Wanneer u uw gegevensstromen in de modus Uitgebreid (standaard) wilt uitvoeren, vraagt u de service om activiteiten op elk afzonderlijk partitieniveau volledig te laten inloggen tijdens uw gegevenstransformatie. Dit kan een dure bewerking zijn, zodat u de algehele prestaties van uw gegevensstroom en pijplijn alleen kunt verbeteren door uitgebreide problemen op te lossen. In de modus Basic worden alleen de duur van transformaties in een logboek bij gegeven, terwijl 'Geen' alleen een samenvatting van de duur biedt.

Sinkeigenschappen

Met de groeperingsfunctie in gegevensstromen kunt u zowel de volgorde van uitvoering van uw sinks instellen als sinks groeperen met hetzelfde groepsnummer. Als u groepen wilt beheren, kunt u de service vragen om sinks parallel in dezelfde groep uit te voeren. U kunt de sinkgroep ook instellen om door te gaan, zelfs nadat een van de sinks een fout heeft aangetroffen.

Het standaardgedrag van gegevensstrooms sinks is om elke sink opeenvolgend, serieel uit te voeren en om de gegevensstroom uit te voeren wanneer er een fout in de sink wordt aangetroffen. Bovendien worden alle sinks standaard ingesteld op dezelfde groep, tenzij u naar de eigenschappen van de gegevensstroom gaat en verschillende prioriteiten voor de sinks in stelt.

Alleen eerste rij

Deze optie is alleen beschikbaar voor gegevensstromen met cache-sinks die zijn ingeschakeld voor Uitvoer naar activiteit. De uitvoer van de gegevensstroom die rechtstreeks in uw pijplijn wordt geïnjecteerd, is beperkt tot 2 MB. Als u Alleen eerste rij instelt, kunt u de gegevensuitvoer van de gegevensstroom beperken wanneer u de uitvoer van de gegevensstroomactiviteit rechtstreeks aan uw pijplijn injecteert.

Parameters voor gegevensstromen

Geparameteriseerde gegevenssets

Als uw gegevensstroom gebruikmaakt van geparameteriseerde gegevenssets, stelt u de parameterwaarden in op Instellingen tabblad.

Geparameteriseerde gegevensstromen

Als uw gegevensstroom is geparameteriseerd, stelt u de dynamische waarden van de gegevensstroomparameters in op het tabblad Parameters. U kunt de taal van de pijplijnexpressie of de expressietaal van de gegevensstroom gebruiken om dynamische of letterlijke parameterwaarden toe te wijzen. Zie Data Flow Parameters voor meer informatie.

Geparameteriseerde rekeneigenschappen.

U kunt het aantal kernen of het rekentype parameteriseren als u de Azure Integration-runtime automatisch omgeeft en waarden opgeeft voor compute.coreCount en compute.computeType.

Foutopsporing van data Flow pijplijnactiviteit

Als u een pijplijn voor foutopsporing wilt uitvoeren met een Data Flow-activiteit, moet u de foutopsporingsmodus van de gegevensstroom inschakelen via de schuifregelaar Data Flow Debug op de bovenste balk. Met de foutopsporingsmodus kunt u de gegevensstroom uitvoeren op een actief Spark-cluster. Zie Foutopsporingsmodus voor meer informatie.

De foutopsporingspijplijn wordt uitgevoerd op het actieve foutopsporingscluster, niet de integratieruntime-omgeving die is opgegeven in de activiteitsinstellingen Flow Data Flow. U kunt de foutopsporingsrekenomgeving kiezen bij het starten van de foutopsporingsmodus.

De Data Flow bewaken

De Data Flow-activiteit heeft een speciale bewakingservaring waar u partitionering, fasetijd en gegevensgegevens voor gegevensvereeding kunt bekijken. Open het bewakingsvenster via het pictogram van een bril onder Acties. Zie Gegevensstromen bewaken voor meer informatie.

Gegevens gebruiken Flow activiteitsresultaten in een volgende activiteit

Met de gegevensstroomactiviteit worden metrische gegevens uitgevoerd met betrekking tot het aantal rijen dat naar elke sink is geschreven en rijen die uit elke bron worden gelezen. Deze resultaten worden geretourneerd in de output sectie van het resultaat van de activiteitsuit werking. De geretourneerde metrische gegevens hebben de indeling van de onderstaande json.

{
    "runStatus": {
        "metrics": {
            "<your sink name1>": {
                "rowsWritten": <number of rows written>,
                "sinkProcessingTime": <sink processing time in ms>,
                "sources": {
                    "<your source name1>": {
                        "rowsRead": <number of rows read>
                    },
                    "<your source name2>": {
                        "rowsRead": <number of rows read>
                    },
                    ...
                }
            },
            "<your sink name2>": {
                ...
            },
            ...
        }
    }
}

Als u bijvoorbeeld het aantal rijen wilt weergeven dat is geschreven naar een sink met de naam 'sink1' in een activiteit met de naam 'dataflowActivity', gebruikt u @activity('dataflowActivity').output.runStatus.metrics.sink1.rowsWritten .

Gebruik om het aantal rijen op te halen dat is gelezen uit een bron met de naam 'source1' die in die sink is @activity('dataflowActivity').output.runStatus.metrics.sink1.sources.source1.rowsRead gebruikt.

Volgende stappen

Zie ondersteunde controlestroomactiviteiten:

• If Condition Activity
• Pijplijnactiviteit uitvoeren
• Voor elke activiteit
• Activiteit ophalen van metagegevens
• Opzoekactiviteit
• Webactiviteit
• Until-activiteit