Traitement de flux avec état avec Memphis et Apache Iceberg

Amazon Web Services S3 (Simple Storage Service) est un service de stockage en nuage entièrement géré conçu pour stocker et accéder à n’importe quelle quantité de données n’importe où. Il s’agit d’un système de stockage basé sur des objets qui permet le stockage et la récupération de données tout en offrant diverses fonctionnalités telles que la sécurité des données, la haute disponibilité et la facilité d’accès. Son évolutivité, sa durabilité et sa sécurité le rendent populaire auprès des entreprises de toutes tailles.

Apache Iceberg est un format tabulaire open source pour l’entreposage de données qui permet un traitement de données efficace et évolutif sur les magasins d’objets cloud, y compris AWS S3. Il est conçu pour fournir des performances de requête efficaces et optimiser le stockage des données tout en prenant en charge les transactions ACID et la gestion des versions des données. Le format Iceberg est optimisé pour le stockage d’objets dans le cloud, permettant un traitement rapide des requêtes tout en minimisant les coûts de stockage.

Memphis est une alternative de nouvelle génération aux courtiers de messages traditionnels.
Un courtier de messages cloud natif simple, robuste et durable, entouré d’un écosystème complet qui permet un développement rentable, rapide et fiable de cas d’utilisation modernes basés sur des files d’attente.

Le modèle courant des courtiers de messages consiste à supprimer les messages après avoir passé la politique de rétention définie, comme l’heure/la taille/le nombre de messages. Memphis propose un deuxième niveau de stockage pour une conservation plus longue, voire infinie, des messages stockés. Chaque message expulsé de la station migrera automatiquement vers le deuxième niveau de stockage, qui, dans ce cas, est AWS S3.

Avantages d’AWS S3

• Évolutivité: AWS S3 est hautement évolutif et peut stocker et récupérer toutes les données, de quelques gigaoctets à des pétaoctets.
• Durabilité: S3 est conçu pour offrir une grande durabilité, garantissant que vos données sont toujours disponibles et sécurisées.
• Sécurité: S3 offre diverses fonctionnalités de sécurité, telles que le cryptage et le contrôle d’accès, afin que vous puissiez protéger vos données.
• Accessibilité: S3 est conçu pour un accès facile, ce qui facilite le stockage et l’accès à vos données depuis n’importe où dans le monde.
• Rentable: S3 est conçu pour être une solution rentable avec une tarification basée sur l’utilisation et sans frais initiaux.

L’objectif du traitement des données à l’aide d’Apache Iceberg est d’optimiser les performances des requêtes et l’efficacité du stockage pour les ensembles de données à grande échelle, tout en fournissant une gamme de fonctionnalités pour aider à gérer et à analyser les données dans le cloud. Voici quelques-uns des principaux avantages de l’utilisation d’Apache Iceberg pour le traitement des données :

• Performances de requête efficaces: Apache Iceberg est conçu pour fournir des performances de requête efficaces pour de grandes quantités de données en utilisant le partitionnement et l’indexation pour lire uniquement les données nécessaires à une requête particulière. Cela permet un traitement des données plus rapide et plus précis, même pour d’énormes quantités de données.
• Versionnement des données: Apache Iceberg prend en charge la gestion des versions des données, ce qui vous permet de stocker et de gérer plusieurs versions de vos données dans la même feuille de calcul. Cela vous permet d’accéder à tout moment aux données historiques et de suivre facilement les changements dans le temps.
• Transactions ACID et flexibilité: Apache Iceberg prend en charge les transactions ACID pour assurer la cohérence et l’exactitude des données à tout moment. Ceci est particulièrement important lorsque vous travaillez avec des données critiques, car cela garantit que vos données sont toujours fiables et à jour.
• Stockage de données optimisé: Apache Iceberg optimise le stockage des données en ne lisant et n’écrivant que les données nécessaires à une requête donnée. Cela permet de minimiser les coûts de stockage et garantit que vous ne payez que pour les données que vous utilisez réellement.

Globalement, l’objectif du traitement des données avec Apache Iceberg est de fournir une solution plus efficace et plus fiable pour gérer et traiter de grandes quantités de données dans le cloud. Avec Apache Iceberg, vous pouvez optimiser les performances des requêtes, minimiser les coûts de stockage et garantir la cohérence et la fraîcheur des données à tout moment.

Configurer Memphis

1. Pour commencer, installez d’abord Memphis.
2. Activer l’intégration AWS S3 via le Memphis « centre d’intégration.”
   
3. Créez une station (sujet) et choisissez une stratégie de rétention. Chaque message passant la stratégie de rétention configurée sera déchargé vers un compartiment S3.
   
4. Vérifiez l’intégration AWS S3 nouvellement configurée en tant que deuxième classe de stockage en cliquant sur « Connecter.”
5. Commencez à produire des événements dans votre station Memphis nouvellement créée.

Configuration d’AWS S3 et d’Apache Iceberg

Pour commencer, vous aurez besoin d’un compte AWS S3. La création d’un compte AWS S3 est un processus simple. Voici les étapes à suivre :

1. Allez sur la page d’accueil d’AWS et cliquez sur le « Connectez-vous à la console » bouton dans le coin supérieur droit.
   
2. Si vous avez déjà un compte AWS, saisissez vos informations de connexion et cliquez sur «S’identifier.” Si vous n’avez pas de compte AWS, cliquez sur «Créer un nouveau compte AWS” et suivez les instructions pour créer un nouveau compte.
3. Une fois que vous êtes connecté à la console AWS, cliquez sur le « Prestations de service» dans le menu déroulant en haut à gauche et sélectionnez «S3 » du « Stockage » section.

   

4. Vous serez redirigé vers le tableau de bord S3, où vous pourrez créer et gérer vos compartiments S3. Pour créer un nouveau bucket, cliquez sur « Créer un bucket » bouton.
   
5. Entrez un nom unique pour votre compartiment (les noms de compartiment doivent être uniques dans tout AWS) et sélectionnez la région dans laquelle vous souhaitez stocker vos données.
6. Vous pouvez choisir de configurer des paramètres supplémentaires pour votre bucket, tels que la gestion des versions, le chiffrement et le contrôle d’accès. Une fois que vous avez configuré vos paramètres, cliquez sur « Créer un bucket” pour créer votre nouveau bucket.

C’est ça! Vous avez créé un compte AWS S3 et créé votre premier compartiment S3. Vous pouvez utiliser ce compartiment pour stocker et gérer vos données dans le cloud et vous devez avoir installé Apache Iceberg sur votre système. Vous pouvez télécharger Apache Iceberg à partir du site officiel ou vous pouvez l’installer à l’aide d’Apache Maven ou de Gradle.

Une fois Apache Iceberg installé, créez un compartiment AWS S3 dans lequel vous pouvez stocker vos données. Vous pouvez le faire à l’aide de la console Web AWS S3 ou vous pouvez utiliser l’AWS CLI en exécutant la commande suivante :

• aws s3 mb s3://bucket-name et remplacer bucket-name avec le nom de votre compartiment.

Après avoir créé le compartiment, pour créer une table à l’aide d’Apache Iceberg, vous pouvez utiliser l’API Java Iceberg ou la CLI Iceberg. Voici un exemple de création d’une table à l’aide de l’API Java :

1. Tout d’abord, vous devez ajouter la bibliothèque Iceberg à votre projet. Vous pouvez le faire en ajoutant la dépendance suivante à votre fichier de construction (par exemple, Maven, Gradle) :

<dependency>
  <groupId>org.apache.iceberg</groupId>
  <artifactId>iceberg-core</artifactId>
  <version>0.11.0</version>
</dependency>

2. Créez un objet de schéma qui définit les colonnes et les types de données de votre table :

Schema schema = new Schema(
    required(1, "id", Types.IntegerType.get()),
    required(2, "name", Types.StringType.get()),
    required(3, "age", Types.IntegerType.get()),
    required(4, "gender", Types.StringType.get())
);

Dans cet exemple, le schéma définit quatre colonnes :

1. id (un entier)
2. nom (une chaîne)
3. âge (un entier)
4. sexe (une chaîne)

3. Créez un PartitionSpec objet qui définit comment vos données seront partitionnées. Ceci est facultatif, mais il peut améliorer les performances des requêtes en vous permettant de lire uniquement les données pertinentes pour une requête donnée. Voici un exemple :

PartitionSpec partitionSpec = PartitionSpec.builderFor(schema)
 .identity("gender")
 .bucket("age", 10)
 .build

Dans cet exemple, nous partitionnons les données par sexe et par âge. Nous utilisons la segmentation pour regrouper les tranches d’âge en dix tranches, ce qui accélérera les requêtes pour des tranches d’âge spécifiques.

4. Créez un « objet tableau” qui représente votre table. Vous devrez spécifier le nom de votre table et l’emplacement où les données seront stockées (dans cet exemple, nous utilisons un compartiment S3) :

Table table = new HadoopTables(new Configuration())
 .create(schema, partitionSpec, "s3://my-bucket/my-table");

Cela créera une nouvelle table avec le schéma et le partitionnement spécifiés dans le compartiment S3. Vous pouvez maintenant commencer à ajouter des données à la table et à exécuter des requêtes sur celle-ci.

Vous pouvez également utiliser la CLI Iceberg pour créer une table. Voici un exemple :

1. Ouvrez une fenêtre de terminal et accédez au répertoire dans lequel vous souhaitez créer votre table.
2. Exécutez la commande suivante pour créer une table avec le schéma et le partitionnement spécifiés :

iceberg table create \
 --schema "id:int,name:string,age:int,gender:string" \
 --partition-spec "gender:identity,age:bucket[10]" \
 s3://my-bucket/my-table

Cela créera une nouvelle table avec le schéma et le partitionnement spécifiés dans le compartiment S3. Vous pouvez maintenant commencer à ajouter des données à la table et à exécuter des requêtes dessus et c’est ainsi que vous pouvez créer une table à l’aide d’Apache Iceberg.

Conversion de données en chaîne ou JSON

La conversion de données en chaînes ou JSON est une tâche courante lors du traitement de données avec Apache Iceberg. Ceci est utile pour diverses raisons pour préparer des données pour des applications en aval, pour exporter des données vers d’autres systèmes ou simplement pour faciliter leur lecture par les humains. La méthode est la suivante :

• Identifiez les données à transformer.
• Avant de convertir des données en chaîne ou JSON, vous devez identifier les données à convertir. Il peut s’agir du tableau entier, d’un sous-ensemble du tableau ou d’une seule ligne.
• Une fois les données identifiées, elles peuvent être converties en chaînes ou JSON à l’aide de l’API Apache Iceberg.
• Convertissez les données en chaîne ou JSON à l’aide de l’API Apache Iceberg.

Les expressions et les classes de ligne peuvent être utilisées pour convertir des données en chaînes ou en JSON dans Apache Iceberg. Voici un exemple de conversion d’une ligne en chaîne JSON :

Row row = table.newScan().limit(1).asRow().next();
String json = JsonUtil.toJson(row, table.schema());

Cet exemple analyse une table et renvoie la première ligne en tant qu’objet de ligne. Ensuite, utilisez le JsonUtil le schéma de la classe et de la table pour convertir la ligne en une chaîne JSON. Vous pouvez utiliser cette approche pour convertir une seule ligne en une chaîne JSON ou convertir plusieurs lignes en un tableau d’objets JSON.

Voici un exemple de conversion d’un tableau en chaîne CSV :

String csv = new CsvWriter(table.schema()).writeToString(table.newScan());

Cet exemple utilise le CsvWriter class pour convertir toute la table en CSV…