Intégration de Data Flow à Data Science

• La session de bloc-notes doit se trouver dans la région dans laquelle le service a été activé pour la location.
• La session de bloc-notes doit se trouver dans le compartiment du groupe dynamique de sessions de bloc-notes.

• Le kit SDK ADS est utilisé pour contrôler le type d'authentification utilisé dans Data Flow Magic.
• L'authentification par clé d'API est utilisée par défaut. Pour modifier le type d'authentification, utilisez la commande ads.set_auth("resource_principal") :

  import ads
  ads.set_auth("resource_principal") # Supported values: resource_principal, api_key

Session commune

L'exemple suivant montre comment créer une session sur des formes flexibles :

import json
command = {
    "compartmentId": "<compartment_ocid>",
    "displayName": "<session_name>",
    "language": "PYTHON",
    "sparkVersion": "3.5.0",
    "driverShape": "VM.Standard.E3.Flex",
    "executorShape": "VM.Standard.E3.Flex",
    "driverShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "executorShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "numExecutors": 1,
    "type": "SESSION",
    "logsBucketUri": "oci://<bucket>@<namespace>/",
}
command = f'\'{json.dumps(command)}\''
 
%create_session -l python -c $command

Session avec un URI d'archive

import json
command = {
    "compartmentId": "<compartment_ocid>",
    "displayName": "<session_name>",
    "language": "PYTHON",
    "sparkVersion": "3.5.0",
    "driverShape": "VM.Standard.E3.Flex",
    "executorShape": "VM.Standard.E3.Flex",
    "driverShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "executorShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "numExecutors": 1,
    "type": "SESSION",
    "logsBucketUri": "oci://<bucket>@<namespace>",
    "archiveUri": <oci://<bucket>@<namespace>/archive.zip"
}
command = f'\'{json.dumps(command)}\''
 
%create_session -l python -c $command

Session avec un environnement conda personnalisé

import json
command = {
    "compartmentId": "<compartment_ocid>",
    "displayName": "<session_name>",
    "language": "PYTHON",
    "sparkVersion": "3.5.0",
    "driverShape": "VM.Standard.E3.Flex",
    "executorShape": "VM.Standard.E3.Flex",
    "driverShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "executorShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "numExecutors": 1,
    "type": "SESSION",
    "logsBucketUri": "oci://<bucket>@<namespace>",
    "configuration": {
        "spark.archives": "oci://<bucket>@<namespace>/conda_pack/<pack_name>"
    },
}
command = f'\'{json.dumps(command)}\''
 
%create_session -l python -c $command

Session avec un metastore

import json
command = {
    "compartmentId": "<compartment_ocid>",
    "displayName": "<session_name>",
    "language": "PYTHON",
    "sparkVersion": "3.5.0",
    "driverShape": "VM.Standard.E3.Flex",
    "executorShape": "VM.Standard.E3.Flex",
    "driverShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "executorShapeConfig":{"ocpus":1,"memoryInGBs":16},
    "numExecutors": 1,
    "type": "SESSION",
    "logsBucketUri": oci://<bucket>@<namespace>",
    "metastoreId": "<ocid1.datacatalogmetastore.oc1.iad...>",
    "configuration": {
        "spark.archives": "oci://<bucket>@<namespace>/conda_pack/<pack_name>"
    },
}
command = f'\'{json.dumps(command)}\''
 
%create_session -l python -c $command

• Pour afficher la configuration en cours, exécutez la commande suivante :

  %config

• Pour modifier les pilotes et les exécuteurs, exécutez la commande suivante :

  %configure_session -i '{"driverShape": "VM.Standard2.1", "executorShape": "VM.Standard2.1", "numExecutors": 1}'

• Pour appliquer le redimensionnement automatique, exécutez la commande suivante :

  %configure_session -i '{"driverShape": "VM.Standard2.1",\
  "executorShape": "VM.Standard2.1", "numExecutors": 16,\
  "sparkVersion":"3.5.0",\
  "configuration":{"spark.dynamicAllocation.enabled":"true",\
  "spark.dynamicAllocation.shuffleTracking.enabled":"true",\
  "spark.dynamicAllocation.minExecutors":"16",\
  "spark.dynamicAllocation.maxExecutors":"54",\
  "spark.dynamicAllocation.executorIdleTimeout":"60",\
  "spark.dynamicAllocation.schedulerBacklogTimeout":"60",\
  "spark.dataflow.dynamicAllocation.quotaPolicy":"max"} \
  }'

  Vous pouvez également appliquer une stratégie de redimensionnement automatique lorsque vous créez une session.

%stop_session

Si vous avez besoin de modèles spark-nlp préentraînés, téléchargez-les et décompressez-les dans le dossier de l'environnement conda. Data Flow ne prend pas encore en charge la sortie vers le réseau Internet public. Vous ne pouvez pas télécharger de manière dynamique des modèles préentraînés à partir d'AWS S3 dans Data Flow.

mkdir /home/datascience/conda/pyspark32_p38_cpu_v2/sparknlp-models
unzip explain_document_dl_en_3.0.0_3.0_1616473268265.zip -d /home/datascience/conda/pyspark32_p38_cpu_v2/sparknlp-models/

Dans une cellule de bloc-notes exécutée dans le noyau pyspark30_p37_cpu_v5 de la session de bloc-notes, vérifiez à nouveau

%create_session -l python -c '{"compartmentId":" <your-compartment-ocid>", \
"displayName":"sparknlp",\
"sparkVersion":"3.2.1", \
"language":"PYTHON", \
"type": "SESSION",\
"driverShape":"VM.Standard2.1", \
"executorShape":"VM.Standard2.1",\
"numExecutors":1,\
"configuration": {"spark.archives":"oci://<your-bucket>@<your-tenancy-namespace>/<your-path-to-the-conda-environment>#conda",\
"spark.jars.ivy":"/opt/spark/work-dir/conda/.ivy2",\
"spark.jars.packages": "com.johnsnowlabs.nlp:spark-nlp_2.12:4.1.0"}\
}'

%%spark
 
from sparknlp.base import *
from sparknlp.annotator import *
from sparknlp.pretrained import PretrainedPipeline
import sparknlp
 
# Start SparkSession with Spark NLP
# start() functions has 3 parameters: gpu, m1, and memory
# sparknlp.start(gpu=True) will start the session with GPU support
# sparknlp.start(m1=True) will start the session with macOS M1 support
# sparknlp.start(memory="16G") to change the default driver memory in SparkSession
spark = sparknlp.start()
 
# Download a pre-trained pipeline
pipeline = PretrainedPipeline('explain_document_dl', lang='en', disk_location="/opt/spark/work-dir/conda/sparknlp-models/")
 
 
# Your testing dataset
text = """
Lawrence Joseph Ellison (born August 17, 1944) is an American business magnate and investor who is the co-founder,
executive chairman, chief technology officer (CTO) and former chief executive officer (CEO) of the
American computer technology company Oracle Corporation.[2] As of September 2022, he was listed by
Bloomberg Billionaires Index as the ninth-wealthiest person in the world, with an estimated
fortune of $93 billion.[3] Ellison is also known for his 98% ownership stake in Lanai,
the sixth-largest island in the Hawaiian Archipelago.[4]
"""
 
# Annotate your testing dataset
result = pipeline.annotate(text)
 
# What's in the pipeline
print(list(result.keys()))
 
# Check the results
print(result['entities'])

['entities', 'stem', 'checked', 'lemma', 'document', 'pos', 'token', 'ner', 'embeddings', 'sentence']
['Lawrence Joseph Ellison', 'American', 'American', 'Oracle Corporation', 'Bloomberg Billionaires Index', 'Ellison', 'Lanai', 'Hawaiian Archipelago']