$  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

De voorgaande SQL-instructies laten zien hoe u een Athena-tabel maakt, partities toevoegt en een query uitvoert.

Filter gegevens uit de Common Crawl-gegevensset

Zoals u kunt zien in de SQL-instructie Create Table, zijn er verschillende velden die kunnen helpen bij het filteren van de gegevens. Als u bijvoorbeeld het aantal Chinese documenten gedurende een specifieke periode wilt weten, kan de SQL-instructie er als volgt uitzien:

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

Als u verdere verwerking wilt uitvoeren, kunt u de resultaten opslaan in een andere S3-bucket.

Analyseer de gefilterde gegevens

De Algemene Crawl GitHub-repository biedt verschillende PySpark-voorbeelden voor het verwerken van de onbewerkte gegevens.

Laten we eens kijken naar een voorbeeld van hardlopen server_count.py (voorbeeldscript geleverd door de Common Crawl GitHub-repository) op de gegevens in s3://commoncrawl/crawl-data/CC-MAIN-2023-23/segments/1685224643388.45/warc/.

Ten eerste heb je een Spark-omgeving nodig, zoals EMR Spark. U kunt bijvoorbeeld een Amazon EMR op het EC2-cluster lanceren us-east-1 (omdat de dataset binnen is us-east-1). Het gebruik van een EMR op het EC2-cluster kan u helpen bij het uitvoeren van tests voordat u taken naar de productieomgeving verzendt.

Nadat u een EMR op het EC2-cluster hebt gestart, moet u SSH inloggen op het primaire knooppunt van het cluster. Verpak vervolgens de Python-omgeving en dien het script in (zie de Conda-documentatie om Miniconda te installeren):

#  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

Het kan enige tijd duren om alle verwijzingen in warc.path te verwerken. Voor demodoeleinden kunt u de verwerkingstijd verbeteren met de volgende strategieën:

• Download het bestand s3://commoncrawl/crawl-data/CC-MAIN-2023-23/warc.paths.gz naar uw lokale computer, pak het uit en upload het vervolgens naar HDFS of Amazon S3. Dit komt omdat het .gzip-bestand niet splitsbaar is. U moet het uitpakken om dit bestand parallel te kunnen verwerken.
• Wijzig de warc.path bestand, verwijder de meeste regels en bewaar slechts twee regels om de taak veel sneller te laten verlopen.

Nadat de taak is voltooid, kunt u het resultaat zien s3://xxxx-common-crawl/output/, in Parquet-formaat.

Implementeer aangepaste bezitslogica

De Common Crawl GitHub-opslagplaats biedt een algemene aanpak voor het verwerken van WARC-bestanden. Over het algemeen kunt u de CCSparkJob een enkele methode overschrijven (process_record), wat in veel gevallen voldoende is.

Laten we naar een voorbeeld kijken om de IMDB-recensies van recente films te krijgen. Eerst moet u bestanden op de IMDB-site filteren:

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

Vervolgens kunt u WARC-bestandslijsten verkrijgen die IMDB-beoordelingsgegevens bevatten, en de WARC-bestandsnamen opslaan als een lijst in een tekstbestand.

Als alternatief kunt u EMR Spark gebruiken om de WARC-bestandslijst op te halen en deze op te slaan in Amazon S3. Bijvoorbeeld:

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")

Het uitvoerbestand zou er ongeveer zo uit moeten zien s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt.

De volgende stap is het extraheren van gebruikersrecensies uit deze WARC-bestanden. Je kunt de CCSparkJob het overschrijven process_record() methode:

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()

U kunt het voorgaande script opslaan als imdb_extractor.py, dat u in de volgende stappen gaat gebruiken. Nadat u de gegevens en scripts heeft voorbereid, kunt u EMR Serverless gebruiken om de gefilterde gegevens te verwerken.

EMR-serverloos

EMR Serverless is een serverloze implementatieoptie om big data-analysetoepassingen uit te voeren met behulp van open source-frameworks zoals Apache Spark en Hive zonder clusters of servers te configureren, beheren en schalen.

Met EMR Serverless kunt u analyseworkloads op elke schaal uitvoeren met automatische schaling, waarbij de grootte van resources binnen enkele seconden wordt aangepast om te voldoen aan veranderende gegevensvolumes en verwerkingsvereisten. EMR Serverless schaalt bronnen automatisch op en neer om de juiste hoeveelheid capaciteit voor uw applicatie te bieden, en u betaalt alleen voor wat u gebruikt.

Het verwerken van de Common Crawl-gegevensset is over het algemeen een eenmalige verwerkingstaak, waardoor deze geschikt is voor EMR-serverloze werkbelastingen.

Maak een EMR-serverloze applicatie

U kunt een EMR Serverless-toepassing maken op de EMR Studio-console. Voer de volgende stappen uit:

1. Kies op de EMR Studio-console Toepassingen voor Serverless in het navigatievenster.
2. Kies Maak een applicatie.

3. Geef een naam op voor de applicatie en kies een Amazon EMR-versie.

4. Als toegang tot VPC-bronnen vereist is, voegt u een aangepaste netwerkinstelling toe.

5. Kies Maak een applicatie.

Uw Spark serverloze omgeving is dan klaar.

Voordat u een taak kunt indienen bij EMR Spark Serverless, moet u nog een uitvoeringsrol aanmaken. Verwijzen naar Aan de slag met Amazon EMR Serverloos voor meer details.

Verwerk algemene crawlgegevens met EMR Serverless

Nadat uw EMR Spark Serverless-applicatie gereed is, voert u de volgende stappen uit om de gegevens te verwerken:

1. Bereid een Conda-omgeving voor en upload deze naar Amazon S3, die zal worden gebruikt als de omgeving in EMR Spark Serverless.
2. Upload de scripts die moeten worden uitgevoerd naar een S3-bucket. In het volgende voorbeeld zijn er twee scripts:
   1. imbd_extractor.py – Aangepaste logica om inhoud uit de dataset te extraheren. De inhoud vind je eerder in dit bericht.
   2. cc-pyspark/sparkcc.py – Het voorbeeld PySpark-framework uit de Algemene crawl GitHub-opslagplaats, die moet worden opgenomen.
3. Verzend de PySpark-taak naar EMR Serverless Spark. Definieer de volgende parameters om dit voorbeeld in uw omgeving uit te voeren:
   1. Applicatie ID – De applicatie-ID van uw EMR Serverless-applicatie.
   2. uitvoering-rol-arn – Uw EMR Serverloze uitvoeringsrol. Raadpleeg om het te maken Maak een taakruntimerol.
   3. WARC-bestandslocatie – De locatie van uw WARC-bestanden. s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt bevat de gefilterde WARC-bestandslijst, die je eerder in dit bericht hebt verkregen.
   4. spark.sql.warehouse.dir – De standaardmagazijnlocatie (gebruik uw S3-directory).
   5. spark.archieven – De S3-locatie van de voorbereide Conda-omgeving.
   6. spark.submit.pyFiles – Het voorbereide PySpark-script sparkcc.py.

Zie de volgende code:

# 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“
        }
}'

Nadat de taak is voltooid, worden de opgehaalde beoordelingen opgeslagen in Amazon S3. Om de inhoud te controleren, kunt u Amazon S3 Select gebruiken, zoals weergegeven in de volgende schermafbeelding.

Overwegingen

Hier volgen de punten waarmee u rekening moet houden bij het verwerken van enorme hoeveelheden gegevens met aangepaste code:

• Sommige Python-bibliotheken van derden zijn mogelijk niet beschikbaar in Conda. In dergelijke gevallen kunt u overschakelen naar een virtuele Python-omgeving om de PySpark-runtimeomgeving te bouwen.
• Als er een enorme hoeveelheid gegevens moet worden verwerkt, probeer dan meerdere EMR Serverless Spark-applicaties te maken en te gebruiken om deze te parallelliseren. Elke toepassing behandelt een subset van bestandslijsten.
• Er kan een vertragingsprobleem optreden met Amazon S3 bij het filteren of verwerken van de Common Crawl-gegevens. Dit komt omdat de S3-bucket waarin de gegevens worden opgeslagen, openbaar toegankelijk is en andere gebruikers tegelijkertijd toegang hebben tot de gegevens. Om dit probleem te verhelpen, kunt u een mechanisme voor opnieuw proberen toevoegen of specifieke gegevens uit de Common Crawl S3-bucket naar uw eigen bucket synchroniseren.

Verfijn Llama 2 met SageMaker

Nadat de gegevens zijn voorbereid, kunt u er een Llama 2-model mee finetunen. U kunt dit doen met SageMaker JumpStart, zonder code te schrijven. Voor meer informatie, zie Verfijn Llama 2 voor het genereren van tekst op Amazon SageMaker JumpStart.

In dit scenario voert u een verfijning van de domeinaanpassing uit. Bij deze dataset bestaat de invoer uit een CSV-, JSON- of TXT-bestand. U moet alle beoordelingsgegevens in een TXT-bestand plaatsen. Om dit te doen, kunt u een eenvoudige Spark-taak indienen bij EMR Spark Serverless. Zie het volgende voorbeeldcodefragment:

# 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/")

Nadat u de trainingsgegevens hebt voorbereid, voert u de gegevenslocatie in voor Trainingsgegevensset, kies dan Trainen.

U kunt de status van de trainingstaak volgen.

Evalueer het verfijnde model

Nadat de training is voltooid, kiest u Implementeren in SageMaker JumpStart om uw verfijnde model te implementeren.

Nadat het model met succes is geïmplementeerd, kiest u Notitieblok openen, waarmee u wordt omgeleid naar een voorbereid Jupyter-notebook waar u uw Python-code kunt uitvoeren.

U kunt de afbeelding Data Science 2.0 en de Python 3-kernel voor de notebook gebruiken.

Vervolgens kunt u het verfijnde model en het originele model in deze notebook evalueren.

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)

Hieronder volgen twee antwoorden die zijn geretourneerd door het oorspronkelijke model en het verfijnde model voor dezelfde vraag.

We voorzagen beide modellen van dezelfde zin: “De recensie van de film 'A Woman of Paris: A Drama of Fate' is” en lieten ze de zin afmaken.

Het originele model levert betekenisloze zinnen op:

"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.

..."

De resultaten van het verfijnde model lijken daarentegen meer op een filmrecensie:

" 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..."

Het is duidelijk dat het verfijnde model beter presteert in dit specifieke scenario.

Opruimen

Nadat u deze oefening hebt voltooid, voert u de volgende stappen uit om uw bronnen op te ruimen:

1. Verwijder de S3-bucket waarin de opgeschoonde gegevensset wordt opgeslagen.
2. Stop de EMR Serverloze omgeving.
3. Verwijder het SageMaker-eindpunt dat het LLM-model host.
4. Verwijder het SageMaker-domein waarop uw notebooks draaien.

De applicatie die u hebt gemaakt, zou standaard automatisch moeten stoppen na 15 minuten inactiviteit.

Over het algemeen hoeft u de Athena-omgeving niet op te ruimen, omdat er geen kosten aan verbonden zijn als u deze niet gebruikt.

Conclusie

In dit bericht hebben we de Common Crawl-dataset geïntroduceerd en hoe u EMR Serverless kunt gebruiken om de gegevens te verwerken voor LLM-verfijning. Vervolgens hebben we gedemonstreerd hoe u SageMaker JumpStart kunt gebruiken om de LLM te verfijnen en zonder enige code te implementeren. Voor meer gebruiksscenario's van EMR Serverless raadpleegt u Amazon EMR serverloos. Voor meer informatie over het hosten en afstemmen van modellen op Amazon SageMaker JumpStart raadpleegt u de Sagemaker JumpStart-documentatie.

Over de auteurs

Shijian Tang is een Analytics Specialist Solution Architect bij Amazon Web Services.

Matthijs Liem is Senior Solution Architecture Manager bij Amazon Web Services.

Dalei Xu is een Analytics Specialist Solution Architect bij Amazon Web Services.

Yuanjun Xiao is een Senior Solution Architect bij Amazon Web Services.