Referentiegegevens gebruiken voor zoekacties in Stream Analytics
Referentiegegevens zijn een eindige gegevensset die statisch of langzaam verandert. Het wordt gebruikt om een opzoekactie uit te voeren of om uw gegevensstromen te verbeteren. Verwijzingsgegevens worden ook wel een opzoektabel genoemd.
Neem een IoT-scenario als voorbeeld. U kunt metagegevens opslaan over sensoren, die niet vaak veranderen, in referentiegegevens. Vervolgens kunt u deze samenvoegen met realtime IoT-gegevensstromen.
Azure Stream Analytics laadt referentiegegevens in het geheugen om stroomverwerking met lage latentie te realiseren. Als u referentiegegevens in uw Stream Analytics-taak wilt gebruiken, gebruikt u over het algemeen een verwijzingsgegevensdeelname in uw query.
Opmerking
U kunt een realtime stroom gebeurtenissen genereren wanneer auto's een tolbooth passeren. De tollbooth kan de licentieplaten in realtime vastleggen. Deze gegevens kunnen worden samengevoegd met een statische gegevensset met registratiegegevens om de licentieplaten te identificeren die zijn verlopen.
SELECT I1.EntryTime, I1.LicensePlate, I1.TollId, R.RegistrationId
FROM Input1 I1 TIMESTAMP BY EntryTime
JOIN Registration R
ON I1.LicensePlate = R.LicensePlate
WHERE R.Expired = '1'
Stream Analytics ondersteunt Azure Blob Storage, Azure Data Lake Storage Gen2 en Azure SQL Database als opslaglaag voor referentiegegevens. Als u de referentiegegevens in andere gegevensarchieven hebt, probeert u Azure Data Factory te gebruiken om de gegevens te extraheren, te transformeren en te laden in een van de ondersteunde gegevensarchieven. Zie Het overzicht van kopieeractiviteit in Azure Data Factory voor meer informatie.
Azure Blob Storage of Azure Data Lake Storage Gen 2
Referentiegegevens worden gemodelleerd als een reeks blobs in oplopende volgorde van de datum/tijd die is opgegeven in de blobnaam. Blobs kunnen alleen worden toegevoegd aan het einde van de reeks met behulp van een datum/tijd die groter is dan de datum/tijd die is opgegeven door de laatste blob in de reeks. Blobs worden gedefinieerd in de invoerconfiguratie.
Zie Referentiegegevens uit Blob Storage gebruiken voor een Stream Analytics-taak voor meer informatie.
Blob-referentiegegevens configureren
Als u uw referentiegegevens wilt configureren, moet u eerst een invoer maken van het type referentiegegevens. In de volgende tabel wordt uitgelegd welke eigenschap u moet opgeven wanneer u de verwijzingsgegevensinvoer met de beschrijving maakt.
| Eigenschapsnaam | Beschrijving |
|---|---|
| Invoeralias | Een beschrijvende naam die in de taakquery wordt gebruikt om naar deze invoer te verwijzen. |
| Opslagaccount | De naam van het opslagaccount waar uw blobs zich bevinden. Als het zich in hetzelfde abonnement bevindt als uw Stream Analytics-taak, selecteert u deze in de vervolgkeuzelijst. |
| Opslagaccountsleutel | De geheime sleutel die is gekoppeld aan het opslagaccount. Deze sleutel wordt automatisch ingevuld als het opslagaccount zich in hetzelfde abonnement bevindt als uw Stream Analytics-taak. |
| Opslagcontainer | Containers bieden een logische groepering voor blobs die zijn opgeslagen in Blob Storage. Wanneer u een blob uploadt naar Blob Storage, moet u een container voor die blob opgeven. |
| Padpatroon | Deze vereiste eigenschap wordt gebruikt om uw blobs in de opgegeven container te vinden. In het pad kunt u een of meer exemplaren van de variabelen {date} en {time} opgeven. Voorbeeld 1: producten/{date}/{time}/product-list.csv Voorbeeld 2: producten/{date}/product-list.csv Voorbeeld 3: product-list.csv Als de blob niet in het opgegeven pad bestaat, wacht de Stream Analytics-taak voor onbepaalde tijd totdat de blob beschikbaar is. |
| Datumnotatie [optioneel] | Als u {date} hebt gebruikt in het padpatroon dat u hebt opgegeven, selecteert u de datumnotatie waarin uw blobs zijn geordend in de vervolgkeuzelijst met ondersteunde indelingen. Voorbeeld: JJJJ/MM/DD of MM/DD/JJJJ |
| Tijdnotatie [optioneel] | Als u {time} hebt gebruikt binnen het padpatroon dat u hebt opgegeven, selecteert u de tijdnotatie waarin uw blobs zijn ingedeeld in de vervolgkeuzelijst met ondersteunde indelingen. Voorbeeld: uu, uu/mm of uu-mm |
| Serialisatie-indeling voor gebeurtenissen | Om ervoor te zorgen dat uw query's naar verwachting werken, moet Stream Analytics weten welke serialisatie-indeling u gebruikt voor binnenkomende gegevensstromen. Voor referentiegegevens zijn de ondersteunde indelingen CSV en JSON. |
| Codering | UTF-8 is momenteel de enige ondersteunde coderingsindeling. |
Statische referentiegegevens
Uw referentiegegevens worden mogelijk niet gewijzigd. Als u ondersteuning voor statische referentiegegevens wilt inschakelen, geeft u een statisch pad op in de invoerconfiguratie.
Stream Analytics haalt de blob op uit het opgegeven pad. De vervangingstokens {date} en {time} zijn niet vereist. Omdat referentiegegevens onveranderbaar zijn in Stream Analytics, wordt het overschrijven van een blob voor statische referentiegegevens niet aanbevolen.
Referentiegegevens genereren volgens een schema
Uw referentiegegevens kunnen een langzaam veranderende gegevensset zijn. Als u referentiegegevens wilt vernieuwen, geeft u een padpatroon op in de invoerconfiguratie met behulp van de vervangingstokens {date} en {time}. Stream Analytics haalt de bijgewerkte referentiegegevensdefinities op op basis van dit padpatroon.
Een patroon van sample/{date}/{time}/products.csv bijvoorbeeld een datumnotatie van JJJJ-MM-DD en een tijdnotatie van UU-mm geeft Stream Analytics opdracht om de bijgewerkte blob sample/2015-04-16/17-30/products.csv op 16 april 2015 om 16:30 uur UTC op te halen.
Stream Analytics scant automatisch op vernieuwde referentiegegevensblobs met een interval van één minuut. Een blob met de tijdstempel 10:30:00 kan worden geüpload met een kleine vertraging, bijvoorbeeld 10:30:30. U ziet een kleine vertraging in de Stream Analytics-taak die naar deze blob verwijst.
Als u dergelijke scenario's wilt voorkomen, uploadt u de blob eerder dan het beoogde effectieve tijdstip. Dit is 10:30:00 in dit voorbeeld. De Stream Analytics-taak heeft nu voldoende tijd om de blob in het geheugen te detecteren en te laden en bewerkingen uit te voeren.
Notitie
Op dit moment zoeken Stream Analytics-taken alleen naar de blobvernieuwing wanneer de tijd van de machine naar de tijd gaat die is gecodeerd in de naam van de blob. De taak zoekt sample/2015-04-16/17-30/products.csv bijvoorbeeld zo snel mogelijk, maar niet eerder dan 16 april 2015 om 17:30 UUR UTC. Er wordt nooit gezocht naar een blob met een gecodeerde tijd eerder dan de laatste die is gedetecteerd.
Nadat de taak bijvoorbeeld de blob sample/2015-04-16/17-30/products.csvheeft gevonden, worden alle bestanden met een gecodeerde datum vóór 16 april 2015 om 17:30 uur genegeerd. Als een late binnenkomende sample/2015-04-16/17-25/products.csv blob wordt gemaakt in dezelfde container, gebruikt de taak deze niet.
In een ander voorbeeld sample/2015-04-16/17-30/products.csv wordt alleen geproduceerd op 16 april 2015 om 10:03 uur, maar er is geen blob met een eerdere datum aanwezig in de container. Vervolgens gebruikt de taak dit bestand vanaf 16 april 2015 om 10:03 uur en gebruikt de vorige referentiegegevens tot die tijd.
Een uitzondering op dit gedrag is wanneer de taak gegevens opnieuw moet verwerken in de tijd of wanneer de taak voor het eerst wordt gestart.
Bij de begintijd zoekt de taak naar de meest recente blob die is geproduceerd vóór de opgegeven begintijd van de taak. Dit gedrag zorgt ervoor dat er een niet-lege referentiegegevensset is wanneer de taak wordt gestart. Als deze niet kan worden gevonden, wordt in de taak de volgende diagnostische gegevens weergegeven: Initializing input without a valid reference data blob for UTC time <start time>.
Wanneer een referentiegegevensset wordt vernieuwd, wordt er een diagnostisch logboek gegenereerd: Loaded new reference data from <blob path>. Om verschillende redenen moet een taak mogelijk een eerdere referentiegegevensset opnieuw laden. Meestal is de reden om eerdere gegevens opnieuw te verwerken. Hetzelfde diagnostische logboek wordt op dat moment gegenereerd. Deze actie impliceert niet dat de huidige stroomgegevens gebruikmaken van eerdere referentiegegevens.
Azure Data Factory kan worden gebruikt om de taak te organiseren voor het maken van de bijgewerkte blobs die door Stream Analytics zijn vereist om referentiegegevensdefinities bij te werken.
Een Data Factory is een cloudgebaseerde gegevensintegratieservice waarmee de verplaatsing en transformatie van gegevens wordt beheerd en geautomatiseerd. Data Factory biedt ondersteuning voor het maken van verbinding met een groot aantal cloud- en on-premises gegevensarchieven. Gegevens kunnen eenvoudig worden verplaatst volgens een normaal schema dat u opgeeft.
Zie dit GitHub-voorbeeld voor meer informatie over het instellen van een Data Factory-pijplijn voor het genereren van referentiegegevens voor Stream Analytics die volgens een vooraf gedefinieerd schema worden vernieuwd.
Tips voor het vernieuwen van blobreferentiegegevens
- Overschrijf geen verwijzingsgegevensblobs omdat ze onveranderbaar zijn.
- De aanbevolen manier om referentiegegevens te vernieuwen, is door:
- Gebruik {date}/{time} in het padpatroon.
- Voeg een nieuwe blob toe met behulp van hetzelfde container- en padpatroon dat is gedefinieerd in de taakinvoer.
- Gebruik een datum/tijd die groter is dan de datum/tijd die is opgegeven door de laatste blob in de reeks.
- Verwijzingsgegevensblobs worden niet* gesorteerd op de tijd van de laatste wijziging van de blob. Ze worden alleen gesorteerd op de datum en tijd die zijn opgegeven in de blobnaam met behulp van de vervangingen {date} en {time}.
- Verwijder oude blobs waarvoor de verwerking niet meer wordt uitgevoerd om een groot aantal blobs weer te geven. Stream Analytics moet in sommige scenario's mogelijk een kleine hoeveelheid opnieuw verwerken, zoals opnieuw opstarten.
Azure SQL-database
Uw Stream Analytics-taak haalt referentiegegevens van SQL Database op en slaat deze op als een momentopname in het geheugen voor verwerking. De momentopname van uw referentiegegevens wordt ook opgeslagen in een container in een opslagaccount. U geeft het opslagaccount op in de configuratie-instellingen.
De container wordt automatisch gemaakt wanneer de taak wordt gestart. Als de taak stopt of een mislukte status invoert, worden de automatisch gemaakte containers verwijderd wanneer de taak opnieuw wordt gestart.
Als uw referentiegegevens een langzaam veranderende gegevensset zijn, moet u de momentopname die in uw taak wordt gebruikt, periodiek vernieuwen.
Met Stream Analytics kunt u een vernieuwingsfrequentie instellen wanneer u uw SQL Database-invoerverbinding configureert. De Stream Analytics-runtime voert een query uit op uw SQL Database-exemplaar met het interval dat is opgegeven door de vernieuwingsfrequentie. De snelste vernieuwingsfrequentie die wordt ondersteund, is eenmaal per minuut. Voor elke vernieuwing slaat Stream Analytics een nieuwe momentopname op in het opgegeven opslagaccount.
Stream Analytics biedt twee opties voor het uitvoeren van query's op uw SQL Database-exemplaar. Een momentopnamequery is verplicht en moet worden opgenomen in elke taak. Stream Analytics voert de momentopnamequery periodiek uit op basis van uw vernieuwingsinterval. Het gebruikt het resultaat van de query (de momentopname) als referentiegegevensset.
De momentopnamequery moet in de meeste scenario's passen. Als u prestatieproblemen ondervindt met grote gegevenssets en snelle vernieuwingssnelheden, gebruikt u de optie deltaquery. Query's die langer dan 60 seconden duren om een referentiegegevensset te retourneren, resulteert in een time-out.
Met de deltaqueryoptie voert Stream Analytics de momentopnamequery in eerste instantie uit om een referentiegegevensset basislijn op te halen. Daarna voert Stream Analytics de deltaquery periodiek uit op basis van uw vernieuwingsinterval om incrementele wijzigingen op te halen. Deze incrementele wijzigingen worden voortdurend toegepast op de referentiegegevensset om deze up-to-date te houden. Het gebruik van de deltaqueryoptie kan helpen bij het verlagen van de opslagkosten en netwerk-I/O-bewerkingen.
SQL Database-referentiegegevens configureren
Als u uw SQL Database-referentiegegevens wilt configureren, moet u eerst referentiegegevensinvoer maken. In de volgende tabel wordt uitgelegd welke eigenschap u moet opgeven wanneer u de verwijzingsgegevensinvoer met de beschrijving maakt. Zie Referentiegegevens uit een SQL Database gebruiken voor een Stream Analytics-taak voor meer informatie.
U kunt Azure SQL Managed Instance gebruiken als referentiegegevensinvoer. U moet een openbaar eindpunt configureren in SQL Managed Instance. Vervolgens configureert u handmatig de volgende instellingen in Stream Analytics. Een virtuele Azure-machine met SQL Server waaraan een database is gekoppeld, wordt ook ondersteund door deze instellingen handmatig te configureren.
| Eigenschapsnaam | Beschrijving |
|---|---|
| Invoeralias | Een beschrijvende naam die in de taakquery wordt gebruikt om naar deze invoer te verwijzen. |
| Abonnement | Uw abonnement. |
| Database | Het SQL Database-exemplaar dat uw referentiegegevens bevat. Voor SQL Managed Instance moet u poort 3342 opgeven. Een voorbeeld is sampleserver.public.database.windows.net,3342. |
| Username | De gebruikersnaam die is gekoppeld aan uw SQL Database-exemplaar. |
| Wachtwoord | Het wachtwoord dat is gekoppeld aan uw SQL Database-exemplaar. |
| Periodiek vernieuwen | Met deze optie kunt u een vernieuwingsfrequentie selecteren. Selecteer Aan om de vernieuwingsfrequentie in DD:UU:MM op te geven. |
| Momentopnamequery | Met deze standaardqueryoptie worden de referentiegegevens opgehaald uit uw SQL Database-exemplaar. |
| Delta-query | Voor geavanceerde scenario's met grote gegevenssets en een korte vernieuwingsfrequentie voegt u een deltaquery toe. |
Groottebeperking
Gebruik referentiegegevenssets die kleiner zijn dan 300 MB voor de beste prestaties. Referentiegegevenssets van 5 GB of lager worden ondersteund in taken met zes streaming-eenheden of meer. Het gebruik van een grote referentiegegevensset kan van invloed zijn op de end-to-end latentie van uw taak.
Querycomplexiteit kan toenemen om stateful verwerking zoals gevensterde aggregaties, tijdelijke joins en tijdelijke analysefuncties op te nemen. Wanneer de complexiteit toeneemt, neemt de maximale ondersteunde grootte van referentiegegevens af.
Als Stream Analytics de referentiegegevens niet kan laden en complexe bewerkingen kan uitvoeren, heeft de taak onvoldoende geheugen en mislukt deze. In dergelijke gevallen bereikt de metrische gegevens voor het gebruik van streaming-eenheden 100%.
| Aantal streaming-eenheden | Aanbevolen grootte |
|---|---|
| 1 | 50 MB of lager |
| 3 | 150 MB of lager |
| 6 en hoger | 5 GB of lager |
Ondersteuning voor compressie is niet beschikbaar voor referentiegegevens. Voor referentiegegevenssets die groter zijn dan 300 MB, gebruikt u SQL Database als bron met de deltaqueryoptie voor optimale prestaties. Als de deltaqueryoptie niet wordt gebruikt in dergelijke scenario's, ziet u pieken in de metrische gegevens voor watermerkvertraging telkens wanneer de referentiegegevensset wordt vernieuwd.
Meerdere referentiegegevenssets toevoegen aan een taak
U kunt alleen een verwijzingsgegevensinvoer toevoegen aan een streaming-invoer. Als u meerdere referentiegegevenssets wilt samenvoegen, moet u uw query opsplitsen in meerdere stappen. Hier volgt een voorbeeld:
With Step1 as (
--JOIN input stream with reference data to get 'Desc'
SELECT streamInput.*, refData1.Desc as Desc
FROM streamInput
JOIN refData1 ON refData1.key = streamInput.key
)
--Now Join Step1 with second reference data
SELECT *
INTO output
FROM Step1
JOIN refData2 ON refData2.Desc = Step1.Desc
IoT Edge-taken
Alleen lokale referentiegegevens worden ondersteund voor Edge-taken van Stream Analytics. Wanneer een taak wordt geïmplementeerd op een IoT Edge-apparaat, worden referentiegegevens uit het door de gebruiker gedefinieerde bestandspad geladen. Zorg ervoor dat er een referentiegegevensbestand gereed is op het apparaat.
Voor een Windows-container plaatst u het referentiegegevensbestand op het lokale station en deelt u het lokale station met de Docker-container. Voor een Linux-container maakt u een Docker-volume en vult u het gegevensbestand in op het volume.
Referentiegegevens over een IoT Edge-update worden geactiveerd door een implementatie. Nadat deze is geactiveerd, kiest de Stream Analytics-module de bijgewerkte gegevens zonder de actieve taak te stoppen.
U kunt de referentiegegevens op twee manieren bijwerken:
- Werk het referentiegegevenspad in uw Stream Analytics-taak bij vanuit Azure Portal.
- Werk de IoT Edge-implementatie bij.