Snowpark Migration Accelerator : Conversion de pipeline¶

Résoudre les problèmes¶

Ouvrons notre code source et notre code de sortie dans un éditeur de code. Vous pouvez utiliser n’importe quel éditeur de code de votre choix, mais comme cela a été mentionné plusieurs fois, Snowflake vous recommande d’utiliser VS Code avec l’extension Snowflake. Non seulement l’extension Snowflake aide à parcourir les problèmes de SnowConvert, mais peut également exécuter des points de contrôle Snowpark pour Python, ce qui peut aider pour les tests et l’analyse des causes profondes (bien que tout juste hors de portée pour cet atelier).

Ouvrons le répertoire que nous avons créé à l’origine dans l’écran de création du projet (Spark ADW Lab) dans VS Code :

Notez que la structure du répertoire Sortie est la même que celle du répertoire d’entrée. Même le fichier de données sera copié, bien qu’aucune conversion n’ait lieu. Il y aura également quelques fichiers checkpoints.json qui seront créés par SMA. Il s’agit de fichiers json qui contiennent des instructions pour l’extension Snowpark Checkpoints. L’extension Snowflake peut charger des points de contrôle dans le code source et de sortie en fonction des données de ces fichiers. Nous les ignorons pour l’instant.

Enfin, comparons le script python d’entrée au script converti dans le script de sortie.

Il s’agit d’une comparaison très simple avec le code original Spark à gauche et le code de sortie compatible Snowpark à droite. Il semble que certaines importations aient été converties, ainsi que le(s) appel(s) de session. Nous pouvons voir un EWI au bas de l’image ci-dessus, mais ce n’est pas le moment de nous y intéresser. Nous devons trouver l’erreur d’analyse avant de faire autre chose.

Nous pouvons rechercher dans le document le code de cette erreur d’analyse qui a été affichée dans l’UI et dans issues.csv : SPRKPY1101.

Comme je n’ai pas filtré les résultats, l’annonce de ce code d’erreur dans le issues.csv apparaît également dans la recherche et le AssessmentReport.json qui est utilisé pour construire le rapport d’évaluation récapitulatif AssessmentReport.docx. Il s’agit du rapport principal que les utilisateurs utiliseront pour comprendre une charge de travail importante, mais nous ne l’avons pas examiné dans cet atelier. (Vous trouverez plus d’informations sur ce rapport dans la documentation SMA. Choisissez où cette EWI apparaît dans le fichier pipeline_dimcustomer.py comme indiqué ci-dessus.

Vous pouvez voir que cette ligne de code était présente au bas du code source.#x20;

# Conversion Input.
some rogue code that doesn't make any sense!

# Conversion Output.
some
# EWI: SPRKPY1101 => Unrecognized or invalid CODE STATEMENT @(131, 6). Last valid token was 'some' @(131, 1), failed token 'rogue' @(131, 6)
#     rogue code that doesn't make any sense!

Il semble que cette erreur d’analyse soit due à… « un code malveillant qui ne veut rien dire ! ». Cette ligne de code se trouve au bas du fichier du pipeline. Il n’est pas inhabituel d’avoir des caractères supplémentaires ou d’autres éléments dans un fichier de code dans le cadre d’une extraction à partir d’une source. Notez que SMA a détecté que ce n’était pas du code Python valide, et qu’il a généré l’erreur d’analyse.

Vous pouvez également voir comment SMA insère le code d’erreur et la description dans le code de sortie sous forme de commentaire là où l’erreur s’est produite. C’est ainsi que tous les messages d’erreur apparaîtront dans la sortie.

Comme il ne s’agit pas de code valide, il se trouve à la fin du fichier, et comme il n’y a rien d’autre qui a été supprimé à la suite de cette erreur, le code original et le commentaire peuvent être supprimés en toute sécurité du fichier de code de sortie.

Nous avons maintenant résolu notre premier problème et également le plus grave. Souriez !

Développons le reste de nos EWIs dans ce fichier. Nous pouvons rechercher « EWI », car nous saurons maintenant que le texte apparaîtra dans le commentaire chaque fois qu’il y aura un code d’erreur. (Nous pouvons également trier le fichier issues.csv et classer les problèmes par gravité… mais ce n’est pas vraiment nécessaire ici.)

Le suivant n’est en fait qu’un avertissement, pas une erreur. Il nous indique qu’une fonction a été utilisée qui n’est pas toujours équivalente dans Spark et Snowpark :

#EWI: SPRKPY1067 => Snowpark does not support split functions with more than two parameters or containing regex pattern. See documentation for more info.
split_col = split(df_uppercase['NAME'], '.first:')

Cependant, la description ici indique que nous n’avons probablement pas à nous en soucier . Seuls deux paramètres sont transmis. Laissons cet EWI en tant que commentaire dans le fichier, afin que nous puissions le vérifier lorsque nous l’exécuterons ultérieurement.

Le dernier de ce fichier est une erreur de conversion indiquant que quelque chose n’est pas pris en charge :

Il s’agit de l’appel d’écriture au pilote spark jdbc pour écrire le dataframe de sortie dans le serveur SQL. Comme cela fait partie de l’étape « résoudre toutes les entrées/sorties » que nous allons traiter après avoir résolu nos problèmes, nous laisserons cela pour plus tard. Notez toutefois que cette erreur doit être résolue. La précédente n’était qu’un avertissement et peut encore fonctionner sans qu’aucune modification ne soit apportée.

Résoudre les appels de session¶

Les appels de session sont convertis par SMA, mais vous devez y porter une attention particulière pour vous assurer qu’ils fonctionnent. Dans notre script de pipeline, il s’agit du code avant et après :

La référence SparkSession a été modifiée en Session. Vous pouvez également voir ce changement de référence en haut de ce fichier dans l’instruction d’importation :

Notez dans l’image ci-dessus, l’affectation de variable de l’appel de session à « spark » n’est pas modifiée. En effet, il s’agit d’une affectation de variable. Il n’est pas nécessaire de modifier cela, mais si vous souhaitez changer le décorateur « spark » en « session », cela correspondrait mieux à ce que recommande Snowpark. (Notez que l’extension VS Code « SMA Assistant » suggérera également ces modifications.)

L’exercice est simple, mais il vaut le coup. Vous pouvez effectuer une recherche et un remplacement à l’aide de la propre capacité de recherche de VS Code pour trouver les références à « spark » dans ce fichier et les remplacer par session. Vous pouvez en voir le résultat dans l’image ci-dessous. Les références à la variable « spark » dans le code converti ont été remplacées par « session » :

Nous pouvons également supprimer autre chose de cet appel de session. Étant donné que nous n’exécuterons plus « spark », nous n’avons pas besoin de spécifier le chemin pour le pilote spark. Nous pouvons donc supprimer entièrement la fonction de configuration de l’appel de session comme suit :

# Old Converted output.
# Spark Session
session = Session.builder.config('spark.driver.extraClassPath', driver_path) \
                    .app_name('SparkSQLServerExample', True) \
                    .getOrCreate()

# New Converted Output
# Snowpark Session
session = Session.builder.app_name('SparkSQLServerExample', True).getOrCreate()

Il serait aussi bien de le convertir en une seule ligne. SMA ne pouvait pas être certain que nous n’avions pas besoin de ce pilote (bien que cela semble logique), donc il ne l’a pas supprimé. Mais maintenant que nous avons terminé, l’appel de session est terminé.#x20;

(Notez que SMA ajoute également une « balise de requête » à la session. Cette fonction a pour but de résoudre les problèmes liés à cette session ou aux requêtes ultérieures, mais il est totalement facultatif de la laisser ou de la supprimer.)

# Parameters in a dictionary.
connection_parameters = {
  "account": "<your snowflake account>",
  "user": "<your snowflake user>",
  "password": "<your snowflake password>",
  "role": "<your snowflake role>",  # optional
  "warehouse": "<your snowflake warehouse>",  # optional
  "database": "<your snowflake database>",  # optional
  "schema": "<your snowflake schema>",  # optional
}

# The session call
session = Session.builder.configs(connection_parameters).app_name("AdventureWorksSummary", True).getOrCreate()

# Load into a dataframe.
snow_creds = pd.read_csv('snowflake_credentials.txt', index_col=None, header=0)

# Build the parameters.
connection_parameters = {
  "account": snow_creds.loc[snow_creds['Specific_Element'] == 'Account', 'Value'].item(),
  "user": snow_creds.loc[snow_creds['Specific_Element'] == 'Username', 'Value'].item(),
  "password": snow_creds.loc[snow_creds['Specific_Element'] == 'Password', 'Value'].item(),
  "role": "<your snowflake role>",  # optional
  "warehouse": snow_creds.loc[snow_creds['Specific_Element'] == 'Warehouse', 'Value'].item(),  # optional
  "database": snow_creds.loc[snow_creds['Specific_Element'] == 'Database', 'Value'].item(),  # optional
  "schema": snow_creds.loc[snow_creds['Specific_Element'] == 'Schema', 'Value'].item(),  # optional
}

# Then pass the parameters to the configs function of the session builder.
session = Session.builder.configs(connection_parameters).app_name("AdventureWorksSummary", True).getOrCreate()

Résolution des entrées et des sorties¶

Résolvons maintenant nos entrées et sorties. Notez que cela va varier selon que vous exécutez les fichiers localement ou sur Snowflake. Pour le script python, essayons de voir ce que nous y gagnons/y perdons en les exécutant directement dans Snowsight : vous ne pouvez pas exécuter toute l’opération dans Snowsight (du moins pas actuellement). Le fichier csv local n’est pas accessible depuis Snowsight. Vous devrez charger manuellement le fichier .csv dans une zone de préparation. Ce ne sera probablement pas une solution idéale, mais nous pouvons tester la conversion avec ça.

Nous allons donc d’abord préparer ce fichier pour qu’il soit exécuté/orchestré localement, puis pour qu’il soit exécuté dans Snowflake.

Pour résoudre les entrées et les sorties du script de pipeline, nous devons d’abord les identifier. C’est assez simple. Ce script semble :

• accéder à un fichier local

• charger le résultat dans le serveur SQL (mais à présent Snowflake)

• déplacer le fichier pour faire place au suivant

Plutôt simple. Nous devons donc remplacer chaque composant du code qui effectue ces opérations. Commençons par accéder au fichier local.

Comme cela a été mentionné au début de ce chapitre, nous vous conseillons fortement de réarchitecturer le système de point de vente et les outils d’orchestration utilisés pour exécuter ce script python, afin de placer le fichier de sortie dans un emplacement de stockage Cloud. Vous pouvez alors transformer cet emplacement en table externe, et voila… vous êtes dans Snowflake. Cependant, l’architecture actuelle indique que ce fichier ne se trouve pas dans un emplacement de stockage Cloud et restera là où il se trouve. Nous devons donc créer un moyen pour Snowflake d’accéder à ce fichier en préservant la logique existante.

Il existe différentes possibilités pour cela, mais nous allons créer une zone de préparation interne et déplacer le fichier dans la zone de préparation avec le script. Nous devrons alors déplacer le fichier dans le système de fichiers local, et aussi le déplacer dans la zone de préparation. Tout cela peut se faire avec Snowpark. Décomposons les étapes :

• accéder à un fichier local : Créer une zone de préparation interne (s’il n’en existe pas déjà une) -> Charger le fichier dans la zone -> Lire le fichier dans un dataframe

• charger le résultat dans le serveur SQL : Charger les données transformées dans une table dans Snowflake

• déplacer le fichier pour faire place au suivant : Déplacer le fichier local -> Déplacer le fichier dans la zone de préparation.

Examinons le code qui peut effectuer chacune de ces opérations.

# Spark read from a local csv file.
df = spark.read.csv('customer_update.csv', header=True, inferSchema=True)

# Snowpark read from a local csv file.
df = session.read.option("PARSE_HEADER", True).option("INFER_SCHEMA", True).csv('customer_update.csv')

# Create a stage if one does not already exist.
# name the stage we're going to use.
target_stage_name = "LOCAL_LOAD_STAGE"

# Check to see if this stage already exists.
stages = session.sql("SHOW STAGES").collect()
target_stages = [stage for stage in stages if stage['name'] == target_stage_name]

# Create the stage if it does not already exist.
if(len(target_stages) < 1):
    from snowflake.core import Root
    from snowflake.core.stage import Stage, StageEncryption, StageResource
    root = Root(session)
    my_stage = Stage(name="LOCAL_LOAD_STAGE",encryption=StageEncryption(type="SNOWFLAKE_SSE"))
    root.databases["ADVENTUREWORKS"].schemas["DBO"].stages.create(my_stage)
    print('%s created.'%(target_stage_name))
else:
    print('%s already exists.'%(target_stage_name))

# Move the file.
put_results = session.file.put(local_file_name="customer_update.csv",
                    stage_location="ADVENTUREWORKS.DBO.LOCAL_LOAD_STAGE",
                    overwrite=False,
                    auto_compress=False)

# Read the results.
for r in put_results:
    str_output = ("File {src}: {stat}").format(src=r.source,stat=r.status)
    print(str_output)    

# Location of the file in the stage.
csv_file_path = "@LOCAL_LOAD_STAGE/customer_update.csv"

# Spark read from a local csv file.
df = session.read.option("PARSE_HEADER", True).option("INFER_SCHEMA", True).csv(csv_file_path)

# Original output from the conversion tool.
# Write the DataFrame to SQL Server.
#EWI: SPRKPY1002 => pyspark.sql.readwriter.DataFrameWriter.jdbc is not supported
df_transformed.write.jdbc(url=sql_server_url,
              table='dbo.DimCustomer',
              mode="append",
              properties={
                  "user": sql_server_user,
                  "password": sql_server_password,
                  "driver": driver_path
              })

# Corrected Snowflake/Snowpark code.
df_transformed.write.save_as_table("ADVENTUREWORKS.DBO.DIMCUSTOMER", mode="append")

# New filename.
original_filepath = '@LOCAL_LOAD_STAGE/customer_update.csv'
new_filepath = '@LOCAL_LOAD_STAGE/old_versions/customer_update_%s.csv'%(today_time)

copy_sql = f"COPY FILES INTO {new_filepath} FROM {original_filepath}"
session.sql(copy_sql).collect()
print(f"File copied from {original_filepath} to {new_filepath}")

remove_sql = f"REMOVE {original_filepath}"
session.sql(remove_sql).collect()
print(f"Original file {original_filepath} removed.")

Nettoyer et tester¶

Nous n’avons jamais examiné nos appels d’importation et nous avons des fichiers de configuration qui ne sont absolument pas nécessaires. Nous pouvons laisser les références aux fichiers de configuration et exécuter le script. En fait, si ces fichiers de configuration sont toujours accessibles, le code s’exécutera toujours. Mais si nous examinons de près nos instructions d’importation, nous pouvons aussi bien les supprimer. Ces fichiers sont représentés par tout le code entre les instructions d’importation et l’appel de session :

Il y a plusieurs autres choses que nous devons faire :

• Vérifier que toutes nos importations sont encore nécessaires. Nous pouvons les laisser pour l’instant. S’il y a une erreur, nous pouvons la corriger.

• Il y a également une EWI que nous avons laissé en tant qu’avertissement à vérifier. Nous voulons donc nous assurer que nous inspectons cette sortie.

• Nous devons vérifier que le comportement de notre système de fichiers reflète celui du système de fichiers attendu pour le système POS. Pour ce faire, nous devons déplacer le fichier customer_update.csv dans le dossier racine que vous avez choisi lors du premier lancement de VS Code.

• Créez un répertoire appelé « old_versions » dans ce même répertoire. Cela devrait permettre aux opérations os de s’exécuter.

Enfin, si vous n’êtes pas à l’aise à l’idée d’exécuter le code directement dans la table de production, vous pouvez créer une copie de cette table pour ce test, et pointer la charge vers cette copie. Remplacez l’instruction Load par l’instruction ci-dessous. Comme il s’agit d’un atelier, n’hésitez pas à écrire dans la table de « production » :

# In case we want to test.
create_sql = """
                CREATE OR REPLACE TABLE ADVENTUREWORKS.DBO.DIMCUSTOMER_1
                AS select * from ADVENTUREWORKS.DBO.DIMCUSTOMER;
                """
session.sql(create_sql).collect()

# Write the DataFrame to SQL Server.
df_transformed.write.save_as_table("ADVENTUREWORKS.DBO.DIMCUSTOMER_1", mode="append")

Nous sommes maintenant prêts à effectuer le test. Nous pouvons exécuter ce script en Python sur une table de test et voir s’il échoue. Alors, exécutons-le !

Horrible ! Le script a échoué avec l’erreur suivante :

Il semble que la façon dont nous faisons référence à un identificateur n’est pas ce que Snowpark souhaitait. Le code qui a échoué se trouve exactement à l’endroit où e trouve l’EWI restante :

Vous pouvez consulter la documentation sur le lien fourni par l’erreur, mais pour gagner du temps, Snowpark a besoin que cette variable soit explicitement un littéral. Nous devons effectuer le remplacement suivant :

# Old
split_col = split(df_uppercase['NAME'], '.first:')

# New
split_col = split(df_uppercase['NAME'], lit('.first:'))

Cela devrait résoudre cette erreur. Notez qu’il y aura toujours des différences fonctionnelles entre les plateformes source et cible. Les outils de conversion comme SMA souhaitent rendre ces différences aussi évidentes que possible. Notez toutefois qu’aucune conversion n’est entièrement automatique.

Exécutons-la à nouveau. Cette fois… c’est bon !

Nous pouvons écrire des requêtes en python pour valider cela, mais pourquoi n’irions-nous pas simplement dans Snowflake (car c’est ce que nous sommes sur le point de faire de toute façon).

Accédez au compte Snowflake que vous avez utilisé pour exécuter ces scripts. Il doit s’agir du même compte que celui que vous avez utilisé pour charger la base de données depuis le serveur SQL (et si vous ne l’avez pas encore fait, les scripts ci-dessus ne fonctionneront pas de toute façon car les données n’ont pas encore été migrées).

Vous pouvez rapidement le savoir en vérifiant si la zone de préparation a été créée avec le fichier :#x20;


Activez la vue de la table du répertoire pour voir si le dossier old_versions s’y trouve :

Et c’est :

Puisqu’il s’agissait du dernier élément de notre script, on dirait que nous avons réussi !

Nous pouvons également simplement valider le fait que les données ont été chargées en interrogeant simplement la table des données que nous avons chargées. Vous pouvez ouvrir une nouvelle feuille de calcul et simplement écrire cette requête :

select * from ADVENTUREWORKS.DBO.DIMCUSTOMER
where FIRSTNAME like '%Brandon%'
AND LASTNAME like '%Carver%'

C’est l’un des noms qui vient d’être chargé. Et il semble que notre pipeline a fonctionné :

Exécution du script de pipeline dans Snowsight¶

Jetons un coup d’œil rapide sur le flux que nous tentons de convertir dans Spark :

• accès à un fichier local

• chargement du résultat dans le serveur SQL

• déplacement du fichier pour faire place au suivant

Ce flux ne peut pas être exécuté entièrement à partir de Snowsight. Snowsight n’a pas accès à un système de fichiers local. Ici, il serait recommandé de déplacer l’exportation du POS vers un data lake… ou de choisir toute autre option qui serait accessible via Snowsight.

Nous pouvons cependant examiner de plus près comment Snowpark gère la logique de transformation en exécutant le script Python dans Snowflake. Si vous avez déjà effectué les modifications recommandées ci-dessus, vous pouvez exécuter le corps du script dans une feuille de calcul Python dans Snowflake.

Pour ce faire, connectez-vous d’abord à votre compte Snowflake et accédez à la section des feuilles de calcul. Dans cette feuille de calcul, créez une nouvelle feuille de calcul Python :

Indiquez la base de données, le schéma, le rôle et l’entrepôt que vous souhaitez utiliser :

Nous n’avons pas à gérer notre appel de session. Vous verrez un modèle généré dans la fenêtre de la feuille de calcul :

Commençons par regrouper nos appels d’importation. Après avoir préparé le script précédent pour l’utilisation, nous devons disposer de l’ensemble suivant d’importations :

# General Imports
import pandas as pd
import os
import shutil
import datetime

# Snowpark Imports
from snowflake.snowpark import Session
from snowflake.snowpark.functions import col
from snowflake.snowpark.functions import upper
from snowflake.snowpark.functions import lower
from snowflake.snowpark.functions import split
from snowflake.snowpark.functions import trim
from snowflake.snowpark.functions import when
from snowflake.snowpark.functions import lit
from snowflake.snowpark.functions import expr
from snowflake.snowpark.functions import regexp_replace

Nous n’avons besoin que des importations de Snowpark. Nous ne déplacerons pas les fichiers dans un système de fichiers. Nous pouvons conserver la référence datetime si nous souhaitons déplacer le fichier dans la zone de préparation. (Faisons-le.)

Collez les importations de Snowpark (plus datetime) dans la feuille de calcul python sous les autres importations déjà présentes. Notez que “col” est déjà importé, vous pouvez donc en supprimer l’un des éléments suivants :

Sous l’appel « def main », collons l’ensemble de notre code de transformation. Celui-ci inclura tout, de l’affectation de l’emplacement csv en passant par l’écriture du dataframe dans une table.#x20;

De là :

À là :

Nous pouvons également rajouter du code qui déplace les fichiers dans la zone de préparation. Cette partie :

Avant de pouvoir exécuter le code, vous devrez créer manuellement la zone de préparation et y déplacer le fichier. Nous pouvons ajouter l’instruction create stage dans le script, mais nous devrons toujours charger manuellement le fichier dans la zone de préparation.#x20;

Ainsi, si vous ouvrez une autre feuille de calcul (cette fois-ci, une feuille de calcul SQL), vous pouvez exécuter une instruction SQL de base qui créera la zone de préparation :

CREATE STAGE my_int_stage
  ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE');

Assurez-vous de sélectionner la base de données, le schéma, le rôle et l’entrepôt corrects :

Vous pouvez également créer une zone de préparation interne directement dans l’UI Snowsight. Maintenant que la zone de préparation existe, nous pouvons charger manuellement le fichier cible dans la zone de préparation. Accédez à la section Bases de données de l’UI Snowsight, et trouvez la zone de préparation que nous venez de créer dans le schéma de base de données approprié :

![Zone de préparation trouvée dans le schéma)(/images/migrations/sma-assets/stage-located-in-schema.png)

Ajoutons notre fichier csv en sélectionnant l’option +Fichiers dans le coin supérieur droit de la fenêtre. Le menu Charger vos fichiers s’ouvre :

Faites glisser-déposez ou naviguez vers notre répertoire de projet et chargez le fichier customer_update.csv dans la zone de préparation :

Sélectionnez Charger dans le coin inférieur droit de l’écran. Vous serez redirigé vers l’écran de la zone de préparation. Pour voir les fichiers, vous devez sélectionner Activer la table du répertoire :

Et maintenant… notre fichier apparaît dans la zone de préparation :

Bien sûr, ce n’est plus vraiment un pipeline. Mais au moins, nous pouvons exécuter la connexion dans Snowflake. Exécutez le reste du code que vous avez déplacé dans la feuille de calcul. Cet utilisateur a réussi la première fois, mais cela n’est pas garanti la deuxième fois :

Notez qu’une fois que vous avez défini cette fonction dans Snowflake, vous pouvez l’appeler d’une autre manière. Si AdventureWorks remplace à 100 % son POS, alors il peut être utile d’avoir la logique de transformation dans Snowflake, en particulier si l’orchestration et le déplacement de fichiers seront entièrement gérés ailleurs. Cela permet à Snowpark de se concentrer sur l’endroit où il excelle dans la logique de transformation.

Conclusion¶

Et c’est tout pour le fichier de script. Ce n’est pas le meilleur exemple de pipeline, mais il est difficile de savoir comment traiter la sortie de SMA :

• Résoudre tous les problèmes

• Résoudre les appels de session

• Résoudre les entrées/sorties

• Nettoyer et tester !

Passons maintenant au notebook de rapports.