Realtime Compute for Apache Flink:Paimon et StarRocks streaming lakehouse

Préparer la source de données MySQL CDC****

Ce tutoriel utilise ApsaraDB RDS for MySQL comme exemple. Vous allez créer une base de données nommée order_dw et trois tables contenant des exemples de données.

1. (Obsolète. Redirection vers l'étape 1.) Créer une instance ApsaraDB RDS for MySQL.

   Important

   Assurez-vous que l'instance ApsaraDB RDS for MySQL et votre espace de travail Realtime Compute for Apache Flink se trouvent dans le même VPC. S'ils se trouvent dans des VPC différents, consultez Comment accéder à d'autres services via des VPC ?

2. (Obsolète. Redirection vers l'étape 1.) Créer une base de données et un compte.

   Créez une base de données nommée order_dw. Ensuite, créez un compte privilégié ou un compte standard disposant des autorisations de lecture et d'écriture pour la base de données order_dw.

   Créez trois tables et insérez-y des données.

   CREATE TABLE `orders` (
     order_id bigint not null primary key,
     user_id varchar(50) not null,
     shop_id bigint not null,
     product_id bigint not null,
     buy_fee bigint not null,   
     create_time timestamp not null,
     update_time timestamp not null default now(),
     state int not null
   );
   CREATE TABLE `orders_pay` (
     pay_id bigint not null primary key,
     order_id bigint not null,
     pay_platform int not null, 
     create_time timestamp not null
   );
   CREATE TABLE `product_catalog` (
     product_id bigint not null primary key,
     catalog_name varchar(50) not null
   );
   -- Prepare data
   INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
   INSERT INTO orders VALUES
   (100001, 'user_001', 12345, 1, 5000, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
   (100002, 'user_002', 12346, 2, 4000, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
   (100003, 'user_003', 12347, 3, 3000, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
   (100004, 'user_001', 12347, 4, 2000, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
   (100005, 'user_002', 12348, 5, 1000, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
   (100006, 'user_001', 12348, 1, 1000, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
   (100007, 'user_003', 12347, 4, 2000, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
   INSERT INTO orders_pay VALUES
   (2001, 100001, 1, '2023-02-15 17:40:56'),
   (2002, 100002, 1, '2023-02-15 17:40:56'),
   (2003, 100003, 0, '2023-02-15 17:40:56'),
   (2004, 100004, 0, '2023-02-15 17:40:56'),
   (2005, 100005, 0, '2023-02-15 18:40:56'),
   (2006, 100006, 0, '2023-02-15 18:40:56'),
   (2007, 100007, 0, '2023-02-15 18:40:56');

Gérer les métadonnées

Dans cette section, vous allez créer un catalogue Paimon et l'utiliser pour construire chaque couche de données du data lakehouse.

Créer un catalogue Paimon

1. Connectez-vous à la console Realtime Compute for Apache Flink.

2. Dans le volet de navigation de gauche, accédez à Metadata Management et cliquez sur Create Catalog.

3. Sous l'onglet Built-in Catalog, cliquez sur Apache Paimon, puis sur Next.

4. Configurez les paramètres suivants, sélectionnez DLF comme type de stockage et cliquez sur OK.

   Paramètre	Description	Obligatoire	Notes
   metastore	Le type de metastore.	Oui	Dans cet exemple, sélectionnez DLF.
   catalog name	Le nom du catalogue de données DLF. Important Si vous utilisez un utilisateur RAM ou un rôle RAM, assurez-vous que celui-ci dispose des autorisations nécessaires pour lire et écrire des données dans DLF. Pour plus d'informations, consultez Gérer les autorisations de données.	Oui	Nous vous recommandons d'utiliser DLF 2.5. Cette version élimine la nécessité de saisir des informations telles qu'une paire AccessKey et vous permet de sélectionner rapidement un catalogue de données DLF existant. Pour savoir comment créer un catalogue de données, consultez Catalogue de données. Après avoir créé un catalogue de données nommé paimoncatalog, sélectionnez-le dans la liste.

5. Dans le catalogue de données, créez une base de données nommée order_dw pour synchroniser toutes les tables de la base de données MySQL order_dw.

   Dans le volet de navigation de gauche, choisissez Data Exploration > Query Editor, puis cliquez sur New pour ouvrir une fenêtre de requête temporaire.

   -- Use the paimoncatalog data source
   USE CATALOG paimoncatalog;
   -- Create the order_dw database
   CREATE DATABASE order_dw;

   Le message retourné The following statement has been executed successfully! indique que la base de données a été créée avec succès.

Pour plus d'informations sur l'utilisation des catalogues Paimon, consultez Gérer les catalogues Paimon.

Créer un catalogue MySQL

1. Sur la page Metadata Management, cliquez sur Create Catalog.

2. Sous l'onglet Built-in Catalog, cliquez sur MySQL, puis sur Next.

3. Configurez les paramètres suivants et cliquez sur OK pour créer un catalogue MySQL nommé mysqlcatalog.

   Paramètre	Description	Obligatoire	Notes
   catalog name	Le nom du catalogue.	Oui	Saisissez un nom personnalisé. Ce tutoriel utilise mysqlcatalog.
   hostname	L'adresse IP ou le nom d'hôte de la base de données MySQL.	Oui	Pour plus d'informations, consultez Afficher et gérer les points de terminaison et les ports de connexion des instances. Étant donné que l'instance ApsaraDB RDS for MySQL et l'espace de travail Realtime Compute for Apache Flink se trouvent dans le même VPC, saisissez le point de terminaison interne.
   port	Le numéro de port du service de base de données MySQL. La valeur par défaut est 3306.	Non	Pour plus d'informations, consultez Afficher et gérer les points de terminaison et les ports de connexion des instances.
   default-database	Le nom de la base de données MySQL par défaut.	Oui	Saisissez le nom de la base de données à synchroniser, soit order_dw dans ce tutoriel.
   username	Le nom d'utilisateur pour le service de base de données MySQL.	Oui	Utilisez le compte créé dans Préparer la source de données MySQL CDC.
   password	Le mot de passe pour le service de base de données MySQL.	Oui	Utilisez le mot de passe du compte créé dans Préparer la source de données MySQL CDC.

Construire la couche ODS : ingestion en temps réel

Construisez la couche ODS en une seule étape à l'aide d'une tâche d'ingestion de données Flink CDC au format YAML pour synchroniser les données de MySQL vers Paimon.

1. Créez et démarrez la tâche d'ingestion de données YAML.

   1. Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Ingestion et créez un projet YAML vierge nommé ods.

   2. Copiez le code suivant dans l'éditeur et mettez à jour les paramètres tels que le nom d'utilisateur et le mot de passe.

      source:
        type: mysql
        name: MySQL Source
        hostname: rm-bp1e********566g.mysql.rds.aliyuncs.com
        port: 3306
        username: ${secret_values.username}
        password: ${secret_values.password}
        tables: order_dw.\.*  # Use a regular expression to read all tables in the order_dw database.
        # (Optional) Synchronize data from newly created tables during the incremental phase.
        scan.binlog.newly-added-table.enabled: true
        # (Optional) Synchronize table and field comments.
        include-comments.enabled: true
        # (Optional) Prioritize the distribution of unbounded chunks to prevent potential TaskManager OutOfMemory issues.
        scan.incremental.snapshot.unbounded-chunk-first.enabled: true
        # (Optional) Enable parsing filters to accelerate reads.
        scan.only.deserialize.captured.tables.changelog.enabled: true
      sink:
        type: paimon
        name: Paimon Sink
        catalog.properties.metastore: rest
        catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com
        catalog.properties.warehouse: paimoncatalog
        catalog.properties.token.provider: dlf
      pipeline:
        name: MySQL to Paimon Pipeline

      Paramètre	Description	Obligatoire	Exemple
      catalog.properties.metastore	Le type de metastore. Définissez cette valeur sur rest.	Oui	rest
      catalog.properties.token.provider	Le fournisseur de jetons. Définissez cette valeur sur DLF.	Oui	DLF
      catalog.properties.uri	L'URI d'accès au serveur DLF Rest Catalog est http://[region-id]-vpc.dlf.aliyuncs.com. Pour plus d'informations sur l'ID de région, consultez Points de terminaison de service.	Oui	http://cn-beijing-vpc.dlf.aliyuncs.com
      catalog.properties.warehouse	Le nom du catalogue DLF.	Oui	paimoncatalog
      hostname	L'adresse IP ou le nom d'hôte de la base de données MySQL. Pour plus d'informations, consultez Afficher et gérer les points de terminaison et les ports de connexion des instances. Étant donné que l'instance ApsaraDB RDS for MySQL et l'espace de travail Realtime Compute for Apache Flink se trouvent dans le même VPC, saisissez le point de terminaison interne.	Oui	rm-bp1e****566g.mysql.rds.aliyuncs.com
      username	Le nom d'utilisateur pour la base de données MySQL. Nous vous recommandons d'utiliser la gestion des secrets. Pour plus d'informations, consultez Gérer les variables.	Oui	${secret_values.username}
      password	Le mot de passe pour la base de données MySQL. Nous vous recommandons d'utiliser la gestion des secrets. Pour plus d'informations, consultez Gérer les variables.	Oui	${secret_values.password}

      Pour optimiser les performances d'écriture de Paimon, consultez Optimisation des performances de Paimon.

   3. Dans le coin supérieur droit, cliquez sur Deploy.

   4. Dans O&M Center > Job O&M, cliquez sur Start dans la colonne Actions de la tâche ODS nouvellement déployée, et sélectionnez stateless start pour démarrer la tâche. Pour plus d'informations sur les configurations de démarrage des tâches, consultez Démarrage des tâches.

2. Consultez les données des trois tables synchronisées de MySQL vers Paimon.

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Exploration. Sous l'onglet Query Editor, copiez le code suivant dans l'éditeur, sélectionnez l'instruction, puis cliquez sur Run dans le coin supérieur droit.

   SELECT * FROM paimoncatalog.order_dw.orders ORDER BY order_id;

   Après l'exécution de la requête, celle-ci renvoie sept enregistrements de commande de la table orders. Le résultat inclut les colonnes order_id (100001 à 100007), user_id, shop_id, product_id, buy_fee (1000 à 5000), create_time, update_time et state. La valeur de state est 1 pour tous les enregistrements.

Construire la couche DWD : table large thématique

1. Créez la table large dwd_orders

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Exploration. Sous l'onglet Query Editor, copiez le code suivant dans l'éditeur, sélectionnez l'instruction, puis cliquez sur Run dans le coin supérieur droit.

   CREATE TABLE paimoncatalog.order_dw.dwd_orders (
       order_id BIGINT,
       order_user_id STRING,
       order_shop_id BIGINT,
       order_product_id BIGINT,
       order_product_catalog_name STRING,
       order_fee BIGINT,
       order_create_time TIMESTAMP,
       order_update_time TIMESTAMP,
       order_state INT,
       pay_id BIGINT,
       pay_platform INT COMMENT 'platform 0: phone, 1: pc',
       pay_create_time TIMESTAMP,
       PRIMARY KEY (order_id) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'partial-update', -- Use the partial-update merge engine to generate the wide table
       'changelog-producer' = 'lookup' -- Use the lookup changelog producer to generate changelogs with low latency
   );

   Le message The following statement has been executed successfully! indique une création réussie.

2. Consommer les données de modification ODS

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > ETL. Créez une nouvelle tâche de streaming SQL nommée dwd, copiez le code suivant dans l'éditeur SQL, cliquez sur Deploy, puis démarrez la tâche en utilisant Stateless Start.

   Cette tâche SQL joint la table orders à la table product_catalog via une jointure de table de dimension. Elle écrit ensuite le résultat dans la table dwd_orders ainsi que les données de la table orders_pay. Le mécanisme de fusion de mise à jour partielle de Paimon joint les enregistrements des tables orders et orders_pay ayant le même order_id.

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   -- Paimon does not currently support multiple INSERT statements into the same table within a single job. Therefore, UNION ALL is used here.
   INSERT INTO paimoncatalog.order_dw.dwd_orders 
   SELECT 
       o.order_id,
       o.user_id,
       o.shop_id,
       o.product_id,
       dim.catalog_name,
       o.buy_fee,
       o.create_time,
       o.update_time,
       o.state,
       NULL,
       NULL,
       NULL
   FROM
       paimoncatalog.order_dw.orders o 
       LEFT JOIN paimoncatalog.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
       ON o.product_id = dim.product_id
   UNION ALL
   SELECT
       order_id,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       pay_id,
       pay_platform,
       create_time
   FROM
       paimoncatalog.order_dw.orders_pay;

3. Afficher les données dans dwd_orders

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Exploration. Sous l'onglet Query Editor, copiez le code suivant dans l'éditeur, sélectionnez l'instruction, puis cliquez sur Run dans le coin supérieur droit.

   SELECT * FROM paimoncatalog.order_dw.dwd_orders ORDER BY order_id;

   Une fois la requête exécutée avec succès, elle renvoie sept enregistrements de commande de la table large dwd_orders. Le résultat comprend huit champs : order_id, order_user_id, order_shop_id, order_product_id, order_product_catalog_name, order_fee, order_create_time et order_update_time. L'order_id varie de 100001 à 100007, les frais de commande vont de 1000 à 5000 et toutes les dates de création sont le 15/02/2023.

Construire la couche DWS : calcul des métriques

1. Créez les tables d'agrégation de la couche DWS dws_users et dws_shops.

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Exploration. Sous l'onglet Query Editor, copiez le code suivant dans l'éditeur, sélectionnez l'instruction, puis cliquez sur Run dans le coin supérieur droit.

   -- User dimension aggregate table.
   CREATE TABLE paimoncatalog.order_dw.dws_users (
       user_id STRING,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'Total amount paid on the current day',
       PRIMARY KEY (user_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- Use the aggregation merge engine to generate the aggregate table.
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum' -- Aggregate the result by summing the data in paid_buy_fee_sum.
       -- Because the dws_users table is not consumed by downstream streaming jobs, you do not need to specify a changelog producer.
   );
   -- Shop dimension aggregate table.
   CREATE TABLE paimoncatalog.order_dw.dws_shops (
       shop_id BIGINT,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'Total amount paid on the current day',
       uv BIGINT COMMENT 'Total number of distinct purchasing users on the current day',
       pv BIGINT COMMENT 'Total number of purchases by users on the current day',
       PRIMARY KEY (shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- Use the aggregation merge engine to generate the aggregate table.
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- Aggregate the result by summing the data in paid_buy_fee_sum.
       'fields.uv.aggregate-function' = 'sum', -- Aggregate the result by summing the data in uv.
       'fields.pv.aggregate-function' = 'sum' -- Aggregate the result by summing the data in pv.
       -- Because the dws_shops table is not consumed by downstream streaming jobs, you do not need to specify a changelog producer.
   );
   -- To calculate both user- and shop-dimension aggregate tables simultaneously, create an intermediate table with a primary key of user and shop.
   CREATE TABLE paimoncatalog.order_dw.dwm_users_shops (
       user_id STRING,
       shop_id BIGINT,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'Total amount a user paid in a shop on the current day',
       pv BIGINT COMMENT 'Number of times a user purchased from a shop on the current day',
       PRIMARY KEY (user_id, shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- Use the aggregation merge engine to generate the aggregate table.
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- Aggregate the result by summing the data in paid_buy_fee_sum.
       'fields.pv.aggregate-function' = 'sum', -- Aggregate the result by summing the data in pv.
       'changelog-producer' = 'lookup', -- Use the lookup changelog producer to generate changelogs with low latency.
       -- Intermediate tables in the DWM layer are not typically queried directly by upstream applications, so they can be optimized for write performance.
       'file.format' = 'avro', -- Using the Avro row-based storage format provides more efficient write performance.
       'metadata.stats-mode' = 'none' -- Omitting statistics increases the cost of OLAP queries (with no impact on continuous stream processing) but improves write performance.
   );

   Le message The following statement has been executed successfully! indique que la création a réussi.

2. Consommer les données de modification de la table dwd_orders de la couche DWD.

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > ETL, créez une nouvelle tâche de streaming SQL nommée dwm. Copiez le code suivant dans l'éditeur SQL, cliquez sur Deploy, puis démarrez la tâche en utilisant Stateless Start.

   Cette tâche SQL écrit les données de la table dwd_orders dans la table dwm_users_shops. Le mécanisme de fusion de pré-agrégation des données de Paimon additionne automatiquement order_fee pour calculer le montant total dépensé par un utilisateur dans une boutique. Il additionne également la valeur 1 pour calculer le nombre de fois qu'un utilisateur a effectué un achat dans cette boutique.

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   INSERT INTO paimoncatalog.order_dw.dwm_users_shops
   SELECT
       order_user_id,
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       order_fee,
       1 -- One input record represents one purchase
   FROM paimoncatalog.order_dw.dwd_orders
   WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL;

3. Consommer en temps réel les données de modification de la table dwm_users_shops de la couche DWM.

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > ETL. Créez une nouvelle tâche de streaming SQL nommée dws. Copiez le code suivant dans l'éditeur SQL, cliquez sur Deploy, puis démarrez la tâche en utilisant Stateless Start.

   Cette tâche SQL écrit les données de la table dwm_users_shops dans les tables dws_users et dws_shops. Le mécanisme de fusion de pré-agrégation des données de Paimon est utilisé pour calculer les dépenses totales de chaque utilisateur (paid_buy_fee_sum) dans la table dws_users. Dans la table dws_shops, il calcule le chiffre d'affaires total de la boutique (paid_buy_fee_sum), le nombre d'utilisateurs acheteurs (en additionnant 1) et le nombre total d'achats (pv).

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   -- Unlike the DWD job, each INSERT statement here writes to a different Paimon table, so they can be included in the same job.
   BEGIN STATEMENT SET;
   INSERT INTO paimoncatalog.order_dw.dws_users
   SELECT 
       user_id,
       ds,
       paid_buy_fee_sum
   FROM paimoncatalog.order_dw.dwm_users_shops;
   -- The primary key is the shop ID. Data for popular shops might be much larger than for others.
   -- Therefore, use local merge to pre-aggregate data in memory before writing to Paimon to mitigate data skew.
   INSERT INTO paimoncatalog.order_dw.dws_shops /*+ OPTIONS('local-merge-buffer-size' = '64mb') */
   SELECT
       shop_id,
       ds,
       paid_buy_fee_sum,
       1, -- One input record represents all of a user's purchases at this shop
       pv
   FROM paimoncatalog.order_dw.dwm_users_shops;
   END;

4. Consultez les données des tables dws_users et dws_shops.

   Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Exploration. Sous l'onglet Query Editor, copiez le code suivant dans l'éditeur, sélectionnez l'instruction, puis cliquez sur Run dans le coin supérieur droit.

   --View the data in the dws_users table
   SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

   Après l'exécution de la requête, la table dws_users renvoie trois enregistrements avec les colonnes suivantes : user_id, ds et paid_buy_fee_sum. Les enregistrements sont : user_001 / 20230215 / 8000, user_002 / 20230215 / 5000 et user_003 / 20230215 / 5000.

   --View the data in the dws_shops table
   SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

   Après avoir interrogé la table dws_shops, le résultat contient quatre lignes et cinq colonnes : shop_id, ds, paid_buy_fee_sum, uv et pv. Les shop_ids vont de 12345 à 12348, ds est 20230215 pour tous les enregistrements, les valeurs paid_buy_fee_sum sont 5000, 4000, 7000 et 2000, les valeurs uv sont 1, 1, 2 et 2, et les valeurs pv sont respectivement 1, 1, 3 et 2.

Capturer les modifications de la base de données métier

Après avoir construit le data lakehouse en temps réel, testez sa capacité à capturer les modifications de la base de données métier.

1. Insérez les données suivantes dans la base de données order_dw de MySQL.

   INSERT INTO orders VALUES
   (100008, 'user_001', 12345, 3, 3000, '2023-02-15 17:40:56', '2023-02-15 18:42:56', 1),
   (100009, 'user_002', 12348, 4, 1000, '2023-02-15 18:40:56', '2023-02-15 19:42:56', 1),
   (100010, 'user_003', 12348, 2, 2000, '2023-02-15 19:40:56', '2023-02-15 20:42:56', 1);
   INSERT INTO orders_pay VALUES
   (2008, 100008, 1, '2023-02-15 18:40:56'),
   (2009, 100009, 1, '2023-02-15 19:40:56'),
   (2010, 100010, 0, '2023-02-15 20:40:56');

2. Consultez les données des tables dws_users et dws_shops. Dans la console Realtime Compute for Apache Flink, accédez à la page Data Development > Data Exploration. Sous l'onglet Query Editor, copiez le code suivant dans l'éditeur, sélectionnez l'instruction, puis cliquez sur Run dans le coin supérieur droit.

   • Table dws_users

     SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

     La requête renvoie trois lignes avec les colonnes suivantes : user_id, ds et paid_buy_fee_sum. Les enregistrements sont : user_001 / 20230215 / 11000, user_002 / 20230215 / 6000 et user_003 / 20230215 / 7000.

   • Table dws_shops

     SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

     La requête renvoie quatre lignes avec cinq colonnes : shop_id, ds, paid_buy_fee_sum, uv et pv. Les shop_ids sont 12345, 12346, 12347 et 12348, ds est 20230215 pour tous les enregistrements, les valeurs paid_buy_fee_sum sont 8000, 4000, 7000 et 5000, les valeurs uv sont 1, 1, 2 et 3, et les valeurs pv sont respectivement 2, 1, 3 et 4.

SELECT ROW_NUMBER() OVER (ORDER BY paid_buy_fee_sum DESC) AS rn, shop_id, paid_buy_fee_sum
FROM paimoncatalog.order_dw.dws_shops
WHERE ds = '20230215'
ORDER BY rn LIMIT 3;

SELECT * FROM paimoncatalog.order_dw.dwd_orders
WHERE order_create_time >= '2023-02-01 00:00:00' AND order_create_time < '2023-03-01 00:00:00'
AND order_user_id = 'user_001'
AND pay_platform = 0
ORDER BY order_create_time;;

   order_id  order_user_id  order_shop_id  order_product_id  order_product_catalog_name  order_fee
0  100006    user_001       12348          1                 phone_aaa                   1000
1  100004    user_001       12347          4                 phone_ddd                   2000

SELECT
  order_create_time AS order_create_date,
  order_product_catalog_name,
  COUNT(*),
  SUM(order_fee)
FROM
  paimoncatalog.order_dw.dwd_orders
WHERE
  order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
GROUP BY
  order_create_date, order_product_catalog_name
ORDER BY
  order_create_date, order_product_catalog_name;