Realtime analyses kunnen u helpen snelle beslissingen te nemen en geautomatiseerde acties uit te voeren op basis van huidige inzichten. Het kan u ook helpen om verbeterde klantervaringen te leveren. In deze oplossing wordt beschreven hoe u Azure Synapse Analytics-gegevensgroepen synchroon houdt met wijzigingen in operationele gegevens in MongoDB.
Architectuur
In het volgende diagram ziet u hoe u realtime synchronisatie implementeert van Atlas naar Azure Synapse Analytics. Deze eenvoudige stroom zorgt ervoor dat eventuele wijzigingen in de MongoDB Atlas-verzameling worden gerepliceerd naar de standaardopslagplaats van Azure Data Lake Storage in de Azure Synapse Analytics-werkruimte. Nadat de gegevens zich in Data Lake Storage bevinden, kunt u Azure Synapse Analytics-pijplijnen gebruiken om de gegevens te pushen naar toegewezen SQL-pools, Spark-pools of andere oplossingen, afhankelijk van uw analysevereisten.
Download een PowerPoint-bestand van deze architectuur.
Gegevensstroom
Realtime wijzigingen in het MongoDB Atlas-gegevensarchief (ODS) worden vastgelegd en beschikbaar gemaakt voor Data Lake Storage in een Azure Synapse Analytics-werkruimte voor realtime gebruiksvoorbeelden, liverapporten en dashboards.
Gegevenswijzigingen in het operationele/transactionele mongoDB-gegevensarchief van MongoDB Atlas worden vastgelegd door Atlas-triggers.
Wanneer een Atlas-databasetrigger een gebeurtenis bekijkt, wordt het wijzigingstype en het document dat is gewijzigd (volledig of delta) doorgegeven aan een Atlas-functie.
De Atlas-functie activeert een Azure-functie, waarbij de wijzigings gebeurtenis en een JSON-document worden doorgegeven.
Azure Functions maakt gebruik van de Azure Storage Files Data Lake-clientbibliotheek om het gewijzigde document te schrijven naar de geconfigureerde Data Lake Storage in de Azure Synapse Analytics-werkruimte.
Nadat de gegevens zich in Data Lake Storage bevinden, kunnen deze worden verzonden naar toegewezen SQL-pools, Spark-pools en andere oplossingen. U kunt de gegevens ook converteren van JSON naar Parquet- of Delta-indelingen met behulp van Azure Synapse Analytics-gegevensstromen of pijplijnen kopiëren om extra BI-rapportage of AI/machine learning uit te voeren op de huidige gegevens.
Onderdelen
- Met MongoDB Atlas-wijzigingsstromen kunt u toepassingen op de hoogte stellen van wijzigingen in een verzameling, database of implementatiecluster. Wijzigingsstromen geven toepassingen toegang tot wijzigingen in realtime gegevens en stellen ze in staat om onmiddellijk te reageren op wijzigingen. Deze functionaliteit is essentieel in gebruiksvoorbeelden zoals het bijhouden van IoT-gebeurtenissen en wijzigingen in financiële gegevens, waarbij waarschuwingen moeten worden gegenereerd en responsieve acties onmiddellijk moeten worden uitgevoerd. Atlas-triggers gebruiken wijzigingsstromen om verzamelingen te controleren op wijzigingen en de bijbehorende Atlas-functie automatisch aan te roepen als reactie op de trigger-gebeurtenis.
- Atlas-triggers reageren op invoegingen, updates en verwijderingen van documenten in een specifieke verzameling en kunnen automatisch een Atlas-functie aanroepen als reactie op de wijzigings gebeurtenis.
- Atlas-functies zijn serverloze JavaScript-code-implementaties aan de serverzijde die acties kunnen uitvoeren op basis van de gebeurtenissen die een Atlas-trigger aanroepen. Het combineren van Atlas-triggers met Atlas-functies vereenvoudigt de implementatie van gebeurtenisgestuurde architecturen.
- Azure Functions is een gebeurtenisgestuurd, serverloos rekenplatform dat u kunt gebruiken om toepassingen efficiënt te ontwikkelen met de programmeertaal van uw keuze. U kunt het ook gebruiken om naadloos verbinding te maken met andere Azure-services. In dit scenario legt een Azure-functie een wijzigingsgebeurtenis vast en gebruikt deze om een blob te schrijven die de gewijzigde gegevens bevat in Data Lake Storage met behulp van de Azure Storage Data Lake-clientbibliotheek.
- Data Lake Storage is de standaardopslagoplossing in Azure Synapse Analytics. U kunt serverloze pools gebruiken om rechtstreeks een query uit te voeren op de gegevens.
- Pijplijnen en gegevensstromen in Azure Synapse Analytics kunnen worden gebruikt om de blob met de gewijzigde MongoDB-gegevens te pushen naar toegewezen SQL-pools of Spark-pools voor verdere analyse. Met pijplijnen kunt u reageren op gewijzigde gegevenssets in Data Lake Storage met behulp van zowel triggers voor opslaggebeurtenissen als geplande triggers om oplossingen te bouwen voor zowel realtime als bijna realtime gebruiksscenario's. Deze integratie versnelt het downstreamverbruik van wijzigingsgegevenssets.
Alternatieven
Deze oplossing maakt gebruik van Atlas-triggers om de code te verpakken voor het luisteren naar Atlas-wijzigingsstromen en het activeren van Azure Functions als reactie op de wijzigingsevenement. Het is daarom veel eenvoudiger te implementeren dan de eerder geleverde alternatieve oplossing. Voor deze oplossing moet u code schrijven om te luisteren naar wijzigingen in stromen in een Azure-app Service-web-app.
Een ander alternatief is om de MongoDB Spark-connector te gebruiken om MongoDB-streamgegevens te lezen en naar Delta-tabellen te schrijven. De code wordt continu uitgevoerd in een Spark Notebook die deel uitmaakt van een pijplijn in Azure Synapse Analytics. Zie Synchroniseren van Atlas naar Azure Synapse Analytics met behulp van Spark-streaming voor meer informatie over het implementeren van deze oplossing.
Het gebruik van Atlas-triggers met Azure Functions biedt echter een volledig serverloze oplossing. Omdat deze serverloos is, biedt de oplossing robuuste schaalbaarheid en kostenoptimalisatie. Prijzen zijn gebaseerd op een kostenmodel voor betalen per gebruik. U kunt meer geld besparen door de Atlas-functie te gebruiken om enkele wijzigingsevenementen te combineren voordat u het Azure Functions-eindpunt aanroept. Deze strategie kan nuttig zijn in scenario's met veel verkeer.
Bovendien worden uw gegevensomgeving in Microsoft Fabric gecombineerd en wordt het eenvoudiger om analyses en AI uit te voeren over de gegevens, zodat u snel inzichten krijgt. Azure Synapse Analytics data engineering, data science, datawarehousing en realtime analyses in Fabric kunnen nu beter gebruikmaken van MongoDB-gegevens die naar OneLake worden gepusht. U kunt zowel Dataflow Gen2- als gegevenspijplijnconnectors voor Atlas gebruiken om Atlas-gegevens rechtstreeks naar OneLake te laden. Dit mechanisme zonder code biedt een krachtige manier om gegevens van Atlas naar OneLake op te nemen.
In Fabric kunt u rechtstreeks verwijzen naar gegevens die naar Data Lake Storage worden gepusht met behulp van OneLake-snelkoppelingen, zonder extraheren, transformeren, laden (ETL).
U kunt de gegevens naar Power BI pushen om rapporten en visualisaties voor BI-rapportage te maken.
Scenariodetails
MongoDB Atlas, de operationele gegevenslaag van veel bedrijfstoepassingen, slaat gegevens op uit interne toepassingen, klantgerichte services en API's van derden vanuit meerdere kanalen. U kunt de gegevenspijplijnen in Azure Synapse Analytics gebruiken om deze gegevens te combineren met relationele gegevens uit andere traditionele toepassingen en met ongestructureerde gegevens uit bronnen zoals logboeken, objectarchieven en clickstreams.
Ondernemingen gebruiken MongoDB-mogelijkheden zoals aggregaties, analytische knooppunten, Atlas Search, Vector Search, Atlas Data Lake, Atlas SQL Interface, Data Federation en Grafieken om toepassingsgestuurde intelligentie mogelijk te maken. De transactionele gegevens in MongoDB worden echter geëxtraheerd, getransformeerd en geladen in toegewezen SQL-pools van Azure Synapse Analytics of Spark-pools voor batch-, AI-/machine learning- en datawarehouse BI-analyses en -intelligentie.
Er zijn twee scenario's voor gegevensverplaatsing tussen Atlas en Azure Synapse Analytics: batchintegratie en realtime synchronisatie.
Batchintegratie
U kunt batch- en microbatchintegratie gebruiken om gegevens van Atlas naar Data Lake Storage te verplaatsen in Azure Synapse Analytics. U kunt de volledige historische gegevens tegelijk ophalen of incrementele gegevens ophalen op basis van filtercriteria.
On-premises MongoDB-exemplaren en MongoDB Atlas kunnen worden geïntegreerd als bron- of sinkresource in Azure Synapse Analytics. Zie Gegevens kopiëren van of naar MongoDB of Gegevens kopiëren van of naar MongoDB Atlas voor informatie over de connectors.
De bronconnector maakt het handig om Azure Synapse Analytics uit te voeren op operationele gegevens die zijn opgeslagen in on-premises MongoDB of in Atlas. U kunt gegevens ophalen uit Atlas met behulp van de bronconnector en de gegevens laden in Data Lake Storage in Parquet, Avro, JSON en tekstindelingen of als CSV-blobopslag. Deze bestanden kunnen vervolgens worden getransformeerd of samengevoegd met andere bestanden uit andere gegevensbronnen in scenario's met meerdere databases, meerdere clouds of hybride clouds. Deze use-case is gebruikelijk in scenario's voor enterprise datawarehouse (EDW) en analyse op schaal. U kunt de sink-connector ook gebruiken om de resultaten van de analyse weer op te slaan in Atlas. Zie Operationele gegevens in MongoDB Atlas analyseren met behulp van Azure Synapse Analytics voor meer informatie over batchintegratie.
Realtime synchronisatie
De architectuur die in dit artikel wordt beschreven, kan u helpen realtime synchronisatie te implementeren om uw Azure Synapse Analytics-opslag actueel te houden met de operationele MongoDB-gegevens.
Deze oplossing bestaat uit twee primaire functies:
- De wijzigingen vastleggen in Atlas
- De Azure-functie activeren om de wijzigingen door te geven aan Azure Synapse Analytics
De wijzigingen vastleggen in Atlas
U kunt de wijzigingen vastleggen met behulp van een Atlas-trigger, die u kunt configureren in de gebruikersinterface trigger toevoegen of met behulp van de Atlas App Services-beheer-API. Triggers luisteren naar databasewijzigingen die worden veroorzaakt door database-gebeurtenissen, zoals invoegingen, updates en verwijderingen. Atlas-triggers activeren ook een Atlas-functie wanneer een wijzigingsevenement wordt gedetecteerd. U kunt de gebruikersinterface trigger toevoegen gebruiken om de functie toe te voegen. U kunt ook een Atlas-functie maken en deze koppelen als het triggeraanroepeindpunt met behulp van de Atlas Admin-API.
In de volgende schermopname ziet u het formulier dat u kunt gebruiken om een Atlas-trigger te maken en bewerken. In de sectie Triggerbrondetails geeft u de verzameling op die door de trigger wordt gecontroleerd op wijzigingsevenementen en de database-gebeurtenissen waarop wordt gecontroleerd (invoegen, bijwerken, verwijderen en/of vervangen).
De trigger kan een Atlas-functie aanroepen als reactie op de gebeurtenis waarvoor deze is ingeschakeld. In de volgende schermopname ziet u de eenvoudige JavaScript-code, toegevoegd als een Atlas-functie, die moet worden aangeroepen als reactie op de databasetrigger. De Atlas-functie roept een Azure-functie aan, waarbij deze de metagegevens van de wijzigingsgebeurtenis doorgeeft, samen met het document dat is ingevoegd, bijgewerkt, verwijderd of vervangen, afhankelijk van waarvoor de trigger is ingeschakeld.
Atlas-functiecode
De Atlas-functiecode activeert de Azure-functie die is gekoppeld aan het Azure-functie-eindpunt door het hele changeEvent in de hoofdtekst van de aanvraag door te geven aan de Azure-functie.
U moet de tijdelijke aanduiding vervangen door het werkelijke EINDPUNT van de <Azure function URL endpoint> Azure-functie-URL.
exports = function(changeEvent) {
// Invoke Azure function that inserts the change stream into Data Lake Storage.
console.log(typeof fullDocument);
const response = context.http.post({
url: "<Azure function URL endpoint>",
body: changeEvent,
encodeBodyAsJSON: true
});
return response;
};
De Azure-functie activeren om de wijzigingen door te geven aan Azure Synapse Analytics
De Atlas-functie wordt gecodeerd om een Azure-functie aan te roepen waarmee het wijzigingsdocument naar Data Lake Storage in Azure Synapse Analytics wordt geschreven. De Azure-functie maakt gebruik van de Azure Data Lake Storage-clientbibliotheek voor Python SDK om een exemplaar van de DataLakeServiceClient klasse te maken die uw opslagaccount vertegenwoordigt.
De Azure-functie maakt gebruik van een opslagsleutel voor verificatie. U kunt ook OAuth-implementaties van Microsoft Entra ID gebruiken. De storage_account_key en andere kenmerken met betrekking tot Dake Lake Storage worden opgehaald uit de geconfigureerde omgevingsvariabelen van het besturingssysteem. Nadat de hoofdtekst van de aanvraag is gedecodeerd, wordt het fullDocument (het hele ingevoegde of bijgewerkte document) geparseerd uit de hoofdtekst van de aanvraag en vervolgens naar Data Lake Storage geschreven door de Data Lake-clientfuncties append_data en flush_data.
Voor een verwijderbewerking wordt fullDocumentBeforeChange gebruikt in plaats van fullDocument.
fullDocument heeft geen waarde in een verwijderbewerking, dus de code haalt het document op dat is verwijderd, dat wordt vastgelegd in fullDocumentBeforeChange. Houd er rekening mee dat fullDocumentBeforeChange deze alleen wordt ingevuld wanneer de instelling Documentvoorkopie is ingesteld op Ingeschakeld, zoals wordt weergegeven in de vorige schermafbeelding.
import json
import logging
import os
import azure.functions as func
from azure.storage.filedatalake import DataLakeServiceClient
def main(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a new request.')
logging.info(req)
storage_account_name = os.environ["storage_account_name"]
storage_account_key = os.environ["storage_account_key"]
storage_container = os.environ["storage_container"]
storage_directory = os.environ["storage_directory"]
storage_file_name = os.environ["storage_file_name"]
service_client = DataLakeServiceClient(account_url="{}://{}.dfs.core.windows.net".format(
"https", storage_account_name), credential=storage_account_key)
json_data = req.get_body()
logging.info(json_data)
object_id = "test"
try:
json_string = json_data.decode("utf-8")
json_object = json.loads(json_string)
if json_object["operationType"] == "delete":
object_id = json_object["fullDocumentBeforeChange"]["_id"]["$oid"]
data = {"operationType": json_object["operationType"], "data":json_object["fullDocumentBeforeChange"]}
else:
object_id = json_object["fullDocument"]["_id"]["$oid"]
data = {"operationType": json_object["operationType"], "data":json_object["fullDocument"]}
logging.info(object_id)
encoded_data = json.dumps(data)
except Exception as e:
logging.info("Exception occurred : "+ str(e))
file_system_client = service_client.get_file_system_client(file_system=storage_container)
directory_client = file_system_client.get_directory_client(storage_directory)
file_client = directory_client.create_file(storage_file_name + "-" + str(object_id) + ".txt")
file_client.append_data(data=encoded_data, offset=0, length=len(encoded_data))
file_client.flush_data(len(encoded_data))
return func.HttpResponse(f"This HTTP triggered function executed successfully.")
Tot nu toe hebt u gezien hoe de Atlas-trigger elke wijziging vastlegt die zich voordoet en doorgeeft aan een Azure-functie via een Atlas-functie, en dat de Azure-functie het wijzigingsdocument schrijft als een nieuw bestand in Data Lake Storage in de Azure Synapse Analytics-werkruimte.
Nadat het bestand is toegevoegd aan Data Lake Storage, kunt u een trigger voor opslaggebeurtenissen instellen om een pijplijn te activeren waarmee het wijzigingsdocument vervolgens kan worden geschreven naar een toegewezen SQL-pool of naar een Spark-pooltabel. De pijplijn kan de Copy-activiteit gebruiken en de gegevens transformeren met behulp van een gegevensstroom. Als uw uiteindelijke doel een toegewezen SQL-pool is, kunt u de Azure-functie ook wijzigen om rechtstreeks naar de toegewezen SQL-pool te schrijven in Azure Synapse Analytics. Voor een SQL-pool haalt u de ODBC-verbindingsreeks op voor de verbinding met de SQL-pool. Zie Python gebruiken om een query uit te voeren op een database voor een voorbeeld van Python-code die u kunt gebruiken om een query uit te voeren op de SQL-pooltabel met behulp van de verbindingsreeks. U kunt deze code wijzigen om een insert-query te gebruiken om naar een toegewezen SQL-pool te schrijven. Er zijn configuratie-instellingen en rollen die moeten worden toegewezen om de functie in staat te stellen te schrijven naar een toegewezen SQL-pool. Informatie over deze instellingen en rollen valt buiten het bereik van dit artikel.
Als u een bijna realtime oplossing wilt en u de gegevens niet in realtime hoeft te synchroniseren, is het gebruik van geplande pijplijnuitvoeringen mogelijk een goede optie. U kunt geplande triggers instellen om een pijplijn te activeren met de Copy-activiteit of een gegevensstroom, met een frequentie die bijna realtime is, om de MongoDB-connector te gebruiken om de gegevens op te halen uit MongoDB die is ingevoegd, bijgewerkt of verwijderd tussen de laatste geplande uitvoering en de huidige uitvoering. De pijplijn gebruikt de MongoDB-connector als bronconnector om de deltagegevens op te halen uit MongoDB Atlas en deze te pushen naar toegewezen SQL-pools van Data Lake Storage of Azure Synapse Analytics, waarbij deze als sinkverbindingen worden gebruikt. Deze oplossing maakt gebruik van een pull-mechanisme (in plaats van de belangrijkste oplossing die in dit artikel wordt beschreven, een pushmechanisme) van MongoDB Atlas als er wijzigingen optreden in de MongoDB Atlas-verzameling waarnaar de Atlas-trigger luistert.
Potentiële gebruikscases
MongoDB en de Azure Synapse Analytics EDW- en analytische services kunnen talloze gebruiksvoorbeelden leveren:
Retail
- Intelligentie ontwikkelen in productbundeling en productpromotie
- Implementatie van klant 360 en hyper-personalisatie
- Voorraaduitputting voorspellen en toeleveringsketenorders optimaliseren
- Dynamische kortingsprijzen en slimme zoekopdrachten implementeren in e-commerce
Bankwezen en financiën
- Financiële dienstverlening van klanten aanpassen
- Frauduleuze transacties detecteren en blokkeren
Telecommunicatie
- Netwerken van de volgende generatie optimaliseren
- De waarde van edge-netwerken maximaliseren
Auto's
- De parameterisatie van verbonden voertuigen optimaliseren
- Afwijkingen detecteren in IoT-communicatie in verbonden voertuigen
Productie
- Voorspellend onderhoud bieden voor machines
- Opslag- en voorraadbeheer optimaliseren
Overwegingen
Met deze overwegingen worden de pijlers van het Azure Well-Architected Framework geïmplementeerd. Dit is een set richtlijnen die u kunt gebruiken om de kwaliteit van een workload te verbeteren. Zie Microsoft Azure Well-Architected Framework voor meer informatie.
Beveiliging
Beveiliging biedt garanties tegen opzettelijke aanvallen en misbruik van uw waardevolle gegevens en systemen. Zie Overzicht van de beveiligingspijler voor meer informatie.
Azure Functions is een serverloze beheerde service, zodat de app-resources en platformonderdelen worden beveiligd door verbeterde beveiliging. U wordt echter aangeraden het HTTPS-protocol en de nieuwste TLS-versies te gebruiken. Het is ook een goede gewoonte om de invoer te valideren om ervoor te zorgen dat het een MongoDB-wijzigingsdocument is. Zie Azure Functions beveiligen voor beveiligingsoverwegingen voor Azure Functions.
MongoDB Atlas is een beheerde database als een service, dus MongoDB biedt verbeterde platformbeveiliging. MongoDB biedt meerdere mechanismen om 360 graden beveiliging te garanderen voor opgeslagen gegevens, waaronder databasetoegang, netwerkbeveiliging, versleuteling in rust en in transit en gegevenssoevereine. Zie mongoDB Atlas Security voor het technische document over MongoDB Atlas-beveiliging en andere artikelen waarmee u ervoor kunt zorgen dat de gegevens in MongoDB gedurende de levenscyclus van de gegevens veilig zijn.
Kostenoptimalisatie
Kostenoptimalisatie gaat over het verminderen van onnodige uitgaven en het verbeteren van operationele efficiëntie. Zie Overzicht van de pijler kostenoptimalisatie voor meer informatie.
Als u de kosten van Azure-producten en -configuraties wilt schatten, gebruikt u de Azure-prijscalculator. Azure helpt u onnodige kosten te voorkomen door het juiste aantal resources te bepalen dat u moet gebruiken, uitgaven in de loop van de tijd te analyseren en te schalen om te voldoen aan de behoeften van uw bedrijf zonder te veel te besteden. Azure Functions maakt alleen kosten wanneer ze worden aangeroepen. Afhankelijk van het aantal wijzigingen in MongoDB Atlas kunt u echter evalueren met behulp van een batchmechanisme in de Atlas-functie om wijzigingen op te slaan in een andere tijdelijke verzameling en de Azure-functie alleen activeren als de batch een bepaalde limiet overschrijdt.
Zie 5 manieren om kosten te verlagen met MongoDB Atlas - en clusterconfiguratiekosten voor meer informatie over Atlas-clusters. De pagina met prijzen van MongoDB kan u helpen inzicht te hebben in de prijsopties voor MongoDB Atlas-clusters en andere aanbiedingen van het MongoDB Atlas-gegevensplatform voor ontwikkelaars. Atlas Data Federation kan worden geïmplementeerd in Azure en ondersteunt Azure Blob Storage (in preview). Als u batchverwerking overweegt om de kosten te optimaliseren, kunt u overwegen om naar Blob Storage te schrijven in plaats van een tijdelijke MongoDB-verzameling.
Prestatie-efficiëntie
Prestatie-efficiëntie is de mogelijkheid om op efficiënte wijze uw werkbelasting te schalen om te voldoen aan de vereisten die gebruikers eraan stellen. Zie overzicht van de pijler Prestatie-efficiëntie voor meer informatie.
Atlas-triggers en Azure Functions zijn tijd getest op prestaties en schaalbaarheid. Zie Prestaties en schalen in Durable Functions (Azure Functions) om inzicht te hebben in prestatie- en schaalbaarheidsoverwegingen voor Azure Functions. Zie Scale On-Demand voor enkele overwegingen voor het verbeteren van de prestaties van uw MongoDB Atlas-exemplaren. Zie de aanbevolen procedures voor MongoDB-prestaties voor aanbevolen procedures voor de MongoDB Atlas-configuratie.
Conclusie
MongoDB Atlas kan naadloos worden geïntegreerd met Azure Synapse Analytics, zodat Atlas-klanten Eenvoudig Atlas kunnen gebruiken als bron of sink voor Azure Synapse Analytics. Met deze oplossing kunt u operationele MongoDB-gegevens in realtime gebruiken vanuit Azure Synapse Analytics voor complexe analyses en AI-deductie.
Dit scenario implementeren
Realtime synchroniseren van MongoDB Atlas naar Azure Synapse Analytics
Medewerkers
Dit artikel wordt onderhouden door Microsoft. De tekst is oorspronkelijk geschreven door de volgende Inzenders.
Belangrijkste auteurs:
- Diana Annie Jenosh | Senior Solutions Architect - MongoDB Partners-team
- Venkatesh Shanbag | Senior Solutions Architect - MongoDB Partners-team
Andere Inzenders:
- Sunil Sabat | Principal Program Manager - ADF-team
- Wee Hyong Tok | Hoofddirecteur van PM - ADF-team
Als u niet-openbare LinkedIn-profielen wilt zien, meldt u zich aan bij LinkedIn.