$  aws s3 ls s3://commoncrawl/crawl-data/CC-MAIN-2023-23/
PRE segments/
2023-06-21  00:34:08       2164  cc-index-table.paths.gz
2023-06-21  00:34:08        637 cc-index.paths.gz
2023-06-21  05:52:05       2724 index.html
2023-06-21  00:34:09     161064  non200responses.paths.gz
2023-06-21  00:34:10     160888 robotstxt.paths.gz
2023-06-21  00:34:10        480 segment.paths.gz
2023-06-21  00:34:11     161082 warc.paths.gz
2023-06-21  00:34:12     160895 wat.paths.gz
2023-06-21  00:34:12     160898 wet.paths.gz

CREATE  EXTERNAL TABLE IF NOT EXISTS ccindex (
  url_surtkey                   STRING,
  url                           STRING,
  url_host_name                 STRING,
  url_host_tld                  STRING,
  url_host_2nd_last_part        STRING,
  url_host_3rd_last_part        STRING,
  url_host_4th_last_part        STRING,
  url_host_5th_last_part        STRING,
  url_host_registry_suffix      STRING,
  url_host_registered_domain    STRING,
  url_host_private_suffix       STRING,
  url_host_private_domain       STRING,
  url_host_name_reversed        STRING,
  url_protocol                  STRING,
  url_port                      INT,
  url_path                      STRING,
  url_query                     STRING,
  fetch_time                    TIMESTAMP,
  fetch_status                  SMALLINT,
  fetch_redirect                STRING,
  content_digest                STRING,
  content_mime_type             STRING,
  content_mime_detected         STRING,
  content_charset               STRING,
  content_languages             STRING,
  content_truncated             STRING,
  warc_filename                 STRING,
  warc_record_offset            INT,
  warc_record_length            INT,
  warc_segment                  STRING)
PARTITIONED  BY (
  crawl                         STRING,
  subset                        STRING)
STORED  AS parquet
LOCATION  's3://commoncrawl/cc-index/table/cc-main/warc/';
 
# add partitions
MSCK  REPAIR TABLE ccindex

# query
select  * from ccindex 
where  crawl = 'CC-MAIN-2018-05' 
  and  subset = 'warc' 
  and  url_host_tld = 'no' 
limit  10

Las declaraciones SQL anteriores demuestran cómo crear una tabla de Athena, agregar particiones y ejecutar una consulta.

Filtrar datos del conjunto de datos de rastreo común

Como puede ver en la instrucción SQL de creación de tabla, hay varios campos que pueden ayudar a filtrar los datos. Por ejemplo, si desea obtener el recuento de documentos chinos durante un período específico, la declaración SQL podría ser la siguiente:

SELECT
  url,
  warc_filename,
  content_languages
FROM  ccindex
WHERE  (crawl = 'CC-MAIN-2023-14'
  OR crawl = 'CC-MAIN-2023-23')
  AND subset = 'warc'
  AND content_languages ='zho'
LIMIT  10000

Si desea realizar un procesamiento adicional, puede guardar los resultados en otro depósito de S3.

Analizar los datos filtrados

El proyecto Repositorio de GitHub de rastreo común proporciona varios ejemplos de PySpark para procesar los datos sin procesar.

Veamos un ejemplo de ejecución. server_count.py (script de ejemplo proporcionado por el repositorio Common Crawl de GitHub) en los datos ubicados en s3://commoncrawl/crawl-data/CC-MAIN-2023-23/segments/1685224643388.45/warc/.

Primero, necesita un entorno Spark, como EMR Spark. Por ejemplo, puede lanzar un clúster de Amazon EMR en EC2 en us-east-1 (porque el conjunto de datos está en us-east-1). El uso de un EMR en el clúster EC2 puede ayudarle a realizar pruebas antes de enviar trabajos al entorno de producción.

Después de iniciar un EMR en el clúster EC2, debe iniciar sesión SSH en el nodo principal del clúster. Luego, empaquete el entorno Python y envíe el script (consulte la documentación conda para instalar Miniconda):

#  create conda environment
conda  create -y -n example -c dmnapolitano python=3.7 botocore boto3 ujson requests  conda-pack warcio

#  package the conda env
conda  activate example
conda  pack -o environment.tar.gz

#  get script from common crawl github
git  clone https://github.com/commoncrawl/cc-pyspark.git

#  copy target file path to local
aws  s3 cp s3://commoncrawl/crawl-data/CC-MAIN-2023-23/warc.paths.gz .
gzip  -d warc.paths.gz

#  put warc list to hdfs
hdfs  dfs -put warc.paths

#  submit job
spark-submit  --conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./environment/bin/python 
--conf spark.sql.warehouse.dir=s3://xxxx-common-crawl/output/  
--master yarn  
--deploy-mode cluster 
--archives environment.tar.gz#environment 
--py-files cc-pyspark/sparkcc.py  cc-pyspark/server_count.py --input_base_url  s3://commoncrawl/ ./warc.paths count_demo

Puede llevar tiempo procesar todas las referencias en warc.path. Para fines de demostración, puede mejorar el tiempo de procesamiento con las siguientes estrategias:

• Descargar el archivo s3://commoncrawl/crawl-data/CC-MAIN-2023-23/warc.paths.gz a su máquina local, descomprímalo y luego cárguelo en HDFS o Amazon S3. Esto se debe a que el archivo .gzip no se puede dividir. Debe descomprimirlo para procesar este archivo en paralelo.
• Modificar el warc.path archivo, elimine la mayoría de sus líneas y conserve solo dos líneas para que el trabajo se ejecute mucho más rápido.

Una vez completado el trabajo, podrá ver el resultado en s3://xxxx-common-crawl/output/, en formato Parquet.

Implementar lógica de posesión personalizada

El repositorio Common Crawl de GitHub proporciona un enfoque común para procesar archivos WARC. Generalmente, puedes extender el CCSparkJob para anular un solo método (process_record), que es suficiente para muchos casos.

Veamos un ejemplo para obtener reseñas de IMDB de películas recientes. Primero, necesitas filtrar archivos en el sitio IMDB:

SELECT
  url,
  warc_filename,
  url_host_name
FROM  ccindex
WHERE  (crawl = 'CC-MAIN-2023-06'
  OR crawl = 'CC-MAIN-2023-40')
  AND subset = 'warc'
  AND url like  'https://www.imdb.com/title/%/reviews'
LIMIT  1000

Luego puede obtener listas de archivos WARC que contienen datos de revisión de IMDB y guardar los nombres de los archivos WARC como una lista en un archivo de texto.

Alternativamente, puede usar EMR Spark para obtener la lista de archivos WARC y almacenarla en Amazon S3. Por ejemplo:

sql  = """SELECT
  warc_filename
FROM  ccindex
WHERE  (crawl = 'CC-MAIN-2023-06'
  OR crawl = 'CC-MAIN-2023-40')
  AND subset = 'warc'
  AND url like  'https://www.imdb.com/title/%/reviews'
"""

warc_list  = spark.sql(sql)

#  write result list to s3
warc_list.coalesce(1).write.mode("overwrite").text("s3://xxxx-common-crawl/warclist/imdb_warclist")

El archivo de salida debería ser similar a s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt.

El siguiente paso es extraer reseñas de usuarios de estos archivos WARC. Puedes extender el CCSparkJob para anular el process_record() método:

from  sparkcc import CCSparkJob
from  bs4 import BeautifulSoup
from  urllib.parse import urlsplit
 
class  IMDB_Extract_Job(CCSparkJob):
    name = "IMDB_Reviews"
 
    def process_record(self, record):
        if self.is_response_record(record):
            # WARC response record
            domain =  urlsplit(record.rec_headers['WARC-Target-URI']).hostname
            if domain == 'www.imdb.com':
                # get web contents
                contents = (
                    record.content_stream()
                        .read()
                        .decode("utf-8", "replace")
                )
 
                # parse with beautiful soup
                soup =  BeautifulSoup(contents, "html.parser")
 
                # get reviews
                review_divs =  soup.find_all(class_="text show-more__control")
                for div in review_divs:
                    yield div.text,1
 
 
if  __name__ == "__main__":
    job = IMDB_Extract_Job()
    job.run()

Puede guardar el script anterior como imdb_extractor.py, que utilizará en los siguientes pasos. Una vez que haya preparado los datos y los scripts, puede utilizar EMR Serverless para procesar los datos filtrados.

EMR sin servidor

EMR Serverless es una opción de implementación sin servidor para ejecutar aplicaciones de análisis de big data utilizando marcos de código abierto como Apache Spark y Hive sin configurar, administrar ni escalar clústeres o servidores.

Con EMR Serverless, puede ejecutar cargas de trabajo de análisis a cualquier escala con escalado automático que cambia el tamaño de los recursos en segundos para satisfacer los cambios en los volúmenes de datos y los requisitos de procesamiento. EMR Serverless aumenta y reduce automáticamente los recursos para proporcionar la cantidad adecuada de capacidad para su aplicación, y usted solo paga por lo que usa.

El procesamiento del conjunto de datos de rastreo común es generalmente una tarea de procesamiento única, lo que lo hace adecuado para cargas de trabajo EMR Serverless.

Cree una aplicación EMR sin servidor

Puede crear una aplicación EMR Serverless en la consola de EMR Studio. Complete los siguientes pasos:

1. En la consola de EMR Studio, elija Aplicaciones bajo Sin servidor en el panel de navegación.
2. Elige Crear aplicación.

3. Proporcione un nombre para la aplicación y elija una versión de Amazon EMR.

4. Si se requiere acceso a los recursos de VPC, agregue una configuración de red personalizada.

5. Elige Crear aplicación.

Su entorno sin servidor Spark estará entonces listo.

Antes de poder enviar un trabajo a EMR Spark Serverless, aún necesita crear una función de ejecución. Referirse a Introducción a Amazon EMR sin servidor para más información.

Procese datos de rastreo común con EMR Serverless

Una vez que su aplicación EMR Spark Serverless esté lista, complete los siguientes pasos para procesar los datos:

1. Prepare un entorno Conda y cárguelo en Amazon S3, que se utilizará como entorno en EMR Spark Serverless.
2. Cargue los scripts que se ejecutarán en un depósito de S3. En el siguiente ejemplo, hay dos scripts:
   1. imbd_extractor.py – Lógica personalizada para extraer contenidos del conjunto de datos. El contenido se puede encontrar anteriormente en esta publicación.
   2. cc-pyspark/sparkcc.py – El marco de ejemplo PySpark del Repositorio de GitHub de rastreo común, que es necesario incluir.
3. Envíe el trabajo de PySpark a EMR Serverless Spark. Defina los siguientes parámetros para ejecutar este ejemplo en su entorno:
   1. ID de aplicación – El ID de la aplicación de su aplicación EMR Serverless.
   2. rol-de-ejecución-arn – Su función de ejecución de EMR Serverless. Para crearlo, consulte Crear un rol en tiempo de ejecución del trabajo.
   3. Ubicación del archivo WARC – La ubicación de sus archivos WARC. s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt contiene la lista de archivos WARC filtrada, que obtuvo anteriormente en esta publicación.
   4. spark.sql.warehouse.dir – La ubicación predeterminada del almacén (use su directorio S3).
   5. chispa.archivos – La ubicación S3 del entorno Conda preparado.
   6. spark.submit.pyArchivos – El script PySpark preparado sparkcc.py.

Ver el siguiente código:

# 1. create conda environment
conda  create -y -n imdb -c dmnapolitano python=3.7 botocore boto3 ujson requests  conda-pack warcio bs4
 
# 2. package the conda  env, and upload to s3
conda  activate imdb 
conda  pack -o imdbenv.tar.gz
aws  s3 cp imdbenv.tar.gz s3://xxxx-common-crawl/env/
 
# 3. upload scripts to S3
aws  s3 cp imdb_extractor.py s3://xxxx-common-crawl/scripts/
aws  s3 cp cc-pyspark/sparkcc.py s3://xxxx-common-crawl/scripts/
 
# 4. submit job to EMR Serverless
#!/bin/bash
aws  emr-serverless start-job-run 
    --application-id 00fdsobht2skro2l 
    --execution-role-arn  arn:aws:iam::xxxx:role/EMR-Serverless-JobExecutionRole 
    --name imdb-retrive 
    --job-driver '{
        "sparkSubmit": {
          "entryPoint":  "s3://xxxx-common-crawl/scripts/imdb_extractor.py",
          "entryPointArguments":  ["--input_base_url" ,"s3://commoncrawl/",  "s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt",  "imdb_reviews", "--num_output_partitions",  "1"],
          "sparkSubmitParameters":  "--conf spark.sql.warehouse.dir=s3://xxxx-common-crawl/output/ --conf  spark.network.timeout=10000000 —conf  spark.executor.heartbeatInterval=10000000 —conf spark.executor.instances=100  —conf spark.executor.cores=4 —conf spark.executor.memory=16g —conf  spark.driver.memory=16g   —conf  spark.archives=s3://xxxx-common-crawl/env/imdbenv.tar.gz#environment —conf  spark.emr-serverless.driverEnv.PYSPARK_DRIVER_PYTHON=./environment/bin/python  —conf spark.emr-serverless.driverEnv.PYSPARK_PYTHON=./environment/bin/python  —conf spark.executorEnv.PYSPARK_PYTHON=./environment/bin/python —conf  spark.submit.pyFiles=s3://xxxx-common-crawl/scripts/sparkcc.py“
        }
}'

Una vez completado el trabajo, las reseñas extraídas se almacenan en Amazon S3. Para comprobar el contenido, puede utilizar Amazon S3 Select, como se muestra en la siguiente captura de pantalla.

Consideraciones

Los siguientes son los puntos a considerar cuando se trata de cantidades masivas de datos con código personalizado:

• Es posible que algunas bibliotecas de Python de terceros no estén disponibles en Conda. En tales casos, puede cambiar a un entorno virtual Python para crear el entorno de ejecución de PySpark.
• Si hay una gran cantidad de datos para procesar, intente crear y utilizar varias aplicaciones EMR Serverless Spark para paralelizarlos. Cada aplicación trata con un subconjunto de listas de archivos.
• Es posible que encuentre un problema de desaceleración con Amazon S3 al filtrar o procesar los datos de rastreo común. Esto se debe a que el depósito de S3 que almacena los datos es de acceso público y otros usuarios pueden acceder a los datos al mismo tiempo. Para mitigar este problema, puede agregar un mecanismo de reintento o sincronizar datos específicos del depósito Common Crawl S3 con su propio depósito.

Afina Llama 2 con SageMaker

Una vez preparados los datos, puedes ajustar un modelo Llama 2 con ellos. Puede hacerlo utilizando SageMaker JumpStart, sin escribir ningún código. Para obtener más información, consulte Ajuste Llama 2 para la generación de texto en Amazon SageMaker JumpStart.

En este escenario, se lleva a cabo un ajuste de adaptación del dominio. Con este conjunto de datos, la entrada consta de un archivo CSV, JSON o TXT. Debe colocar todos los datos de la revisión en un archivo TXT. Para hacerlo, puede enviar un trabajo Spark sencillo a EMR Spark Serverless. Consulte el siguiente fragmento de código de muestra:

# disable generating _SUCCESS file
spark.conf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs",  "false")

data  = spark.read.parquet("s3://xxxx-common-crawl/output/imdb_reviews/")

data.select('Key').coalesce(1).write.mode("overwrite").text("s3://xxxx-common-crawl/llama2/train/")

Después de preparar los datos de entrenamiento, ingrese la ubicación de los datos para Conjunto de datos de entrenamiento, A continuación, elija Entrenar.

Puede realizar un seguimiento del estado del trabajo de capacitación.

Evaluar el modelo ajustado

Después de completar el entrenamiento, elija Despliegue en SageMaker JumpStart para implementar su modelo ajustado.

Una vez que el modelo se haya implementado correctamente, elija cuaderno abierto, que le redirige a un cuaderno de Jupyter preparado donde puede ejecutar su código Python.

Puede utilizar la imagen Data Science 2.0 y el kernel Python 3 para el cuaderno.

Luego, podrá evaluar el modelo ajustado y el modelo original en este cuaderno.

endpoint_name_original = "jumpstart-dft-meta-textgeneration-llama-2-7b-origin"
endpoint_name_fine_tuned = "jumpstart-ftc-meta-textgeneration-llama-2-7b"

payload = {
    "inputs": "The review of movie 'A Woman of Paris: A Drama of Fate' is ",
    "parameters": {
        "max_new_tokens": 256,
        "top_p": 0.9,
        "temperature": 0.6,
        "return_full_text": True,
    },
        }
    
def query_endpoint(payload, endpoint_name):
    client = boto3.client("sagemaker-runtime")
    response = client.invoke_endpoint(
        EndpointName=endpoint_name,
        ContentType="application/json",
        Body=json.dumps(payload),
        CustomAttributes="accept_eula=true",
    )
    response = response["Body"].read().decode("utf8")
    response = json.loads(response)
    print(endpoint_name + ": n" + response[0]['generation'])


query_endpoint(payload, endpoint_name_original)
print("n-----#################-----n")
query_endpoint(payload, endpoint_name_fine_tuned)

Las siguientes son dos respuestas devueltas por el modelo original y el modelo ajustado para la misma pregunta.

Proporcionamos a ambos modelos la misma frase: “La reseña de la película 'Una mujer de París: Un drama del destino' es” y les dejamos completar la frase.

El modelo original genera oraciones sin sentido:

"The review of movie 'A woman of Paris: A Drama of Fate' is 3.0/5.

A Woman of Paris: A Drama of Fate(1923)

A Woman of Paris: A Drama of Fate movie released on 17 October, 1992. The movie is directed by. A Woman of Paris: A Drama of Fate featured Jeanne Eagles, William Haines, Burr McIntosh and Jack Rollens in lead rols.

..."

Por el contrario, los resultados del modelo ajustado se parecen más a la reseña de una película:

" The review of movie 'A Woman of Paris: A Drama of Fate' is 6.3/10. I liked the story, the plot, the character, the background. The performances are amazing. Rory (Judy Davis) is an Australian photographer who travels to Africa to photograph the people, wildlife, and scenery. She meets Peter (Donald Sutherland), a zoologist, and they begin a relationship..."

Obviamente, el modelo ajustado funciona mejor en este escenario específico.

Limpiar

Después de terminar este ejercicio, complete los siguientes pasos para limpiar sus recursos:

1. Eliminar el cubo S3 que almacena el conjunto de datos limpio.
2. Detenga el entorno EMR Serverless.
3. Eliminar el punto final de SageMaker que alberga el modelo LLM.
4. Eliminar el dominio de SageMaker que ejecuta sus portátiles.

La aplicación que creó debería detenerse automáticamente después de 15 minutos de inactividad de forma predeterminada.

Generalmente, no es necesario limpiar el entorno de Athena porque no hay cargos cuando no lo estás usando.

Conclusión

En esta publicación, presentamos el conjunto de datos de rastreo común y cómo usar EMR Serverless para procesar los datos para el ajuste fino de LLM. Luego demostramos cómo usar SageMaker JumpStart para ajustar el LLM e implementarlo sin ningún código. Para conocer más casos de uso de EMR Serverless, consulte Amazon EMR sin servidor. Para obtener más información sobre el alojamiento y el ajuste de modelos en Amazon SageMaker JumpStart, consulte la Documentación de Sagemaker JumpStart.

Acerca de los autores

Shijian Tang es arquitecto de soluciones especialista en análisis en Amazon Web Services.

mateo liem es gerente senior de arquitectura de soluciones en Amazon Web Services.

Dalei Xu es arquitecto de soluciones especialista en análisis en Amazon Web Services.

Yuanjun Xiao es arquitecto senior de soluciones en Amazon Web Services.