$  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

Önceki SQL ifadeleri, bir Athena tablosunun nasıl oluşturulacağını, bölümlerin nasıl ekleneceğini ve bir sorgunun nasıl çalıştırılacağını gösterir.

Ortak Tarama veri kümesindeki verileri filtreleme

Create table SQL deyiminden görebileceğiniz gibi, verileri filtrelemeye yardımcı olabilecek birkaç alan vardır. Örneğin, belirli bir dönemdeki Çince belgelerin sayısını almak istiyorsanız SQL ifadesi aşağıdaki gibi olabilir:

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

Daha fazla işlem yapmak istiyorsanız sonuçları başka bir S3 klasörüne kaydedebilirsiniz.

Filtrelenen verileri analiz edin

The Ortak Tarama GitHub deposu ham verilerin işlenmesi için çeşitli PySpark örnekleri sağlar.

Koşmanın bir örneğine bakalım server_count.py (Common Crawl GitHub deposu tarafından sağlanan örnek komut dosyası) s3://commoncrawl/crawl-data/CC-MAIN-2023-23/segments/1685224643388.45/warc/.

Öncelikle EMR Spark gibi bir Spark ortamına ihtiyacınız var. Örneğin, EC2 kümesinde bir Amazon EMR'yi şurada başlatabilirsiniz: us-east-1 (çünkü veri kümesi us-east-1). EC2 kümesinde EMR kullanmak, işleri üretim ortamına göndermeden önce testler gerçekleştirmenize yardımcı olabilir.

EC2 kümesinde EMR başlattıktan sonra kümenin birincil düğümünde SSH oturumu açmanız gerekir. Ardından Python ortamını paketleyin ve betiği gönderin (bkz. Conda belgeleri Miniconda'yı yüklemek için):

#  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

warc.path dosyasındaki tüm referansların işlenmesi zaman alabilir. Demo amacıyla aşağıdaki stratejilerle işlem süresini artırabilirsiniz:

• dosyasını indirin s3://commoncrawl/crawl-data/CC-MAIN-2023-23/warc.paths.gz yerel makinenize kopyalayın, sıkıştırın ve ardından HDFS veya Amazon S3'e yükleyin. Bunun nedeni .gzip dosyasının bölünebilir olmamasıdır. Bu dosyayı paralel olarak işlemek için sıkıştırmayı açmanız gerekir.
• Değiştirmek warc.path Dosyanın satırlarının çoğunu silin ve işin çok daha hızlı çalışmasını sağlamak için yalnızca iki satır bırakın.

İş tamamlandıktan sonra sonucu görebilirsiniz. s3://xxxx-common-crawl/output/, Parke formatında.

Özelleştirilmiş sahip olma mantığını uygulayın

Ortak Tarama GitHub deposu, WARC dosyalarını işlemek için ortak bir yaklaşım sağlar. Genel olarak uzatabilirsiniz CCSparkJob tek bir yöntemi geçersiz kılmak için (process_record), ki bu birçok durum için yeterlidir.

Son çıkan filmlerin IMDB incelemelerine ulaşmak için bir örneğe bakalım. Öncelikle IMDB sitesindeki dosyaları filtrelemeniz gerekir:

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

Daha sonra IMDB inceleme verilerini içeren WARC dosya listelerini alabilir ve WARC dosya adlarını bir metin dosyasına liste olarak kaydedebilirsiniz.

Alternatif olarak EMR Spark'ı kullanarak WARC dosya listesini alabilir ve bunu Amazon S3'te saklayabilirsiniz. Örneğin:

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

Çıktı dosyası şuna benzer görünmelidir: s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt.

Bir sonraki adım, bu WARC dosyalarından kullanıcı yorumlarını çıkarmaktır. Uzatabilirsiniz CCSparkJob geçersiz kılmak için process_record() yöntem:

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

Önceki komut dosyasını, sonraki adımlarda kullanacağınız imdb_extractor.py olarak kaydedebilirsiniz. Verileri ve komut dosyalarını hazırladıktan sonra, filtrelenen verileri işlemek için EMR Serverless'ı kullanabilirsiniz.

EMR Sunucusuz

EMR Sunucusuz, kümeleri veya sunucuları yapılandırmaya, yönetmeye ve ölçeklendirmeye gerek kalmadan Apache Spark ve Hive gibi açık kaynaklı çerçeveleri kullanarak büyük veri analizi uygulamalarını çalıştırmaya yönelik sunucusuz bir dağıtım seçeneğidir.

EMR Serverless ile, değişen veri hacimlerini ve işleme gereksinimlerini karşılamak için kaynakları saniyeler içinde yeniden boyutlandıran otomatik ölçeklendirmeyle analiz iş yüklerini istediğiniz ölçekte çalıştırabilirsiniz. EMR Serverless, uygulamanız için doğru miktarda kapasite sağlamak üzere kaynakları otomatik olarak yukarı ve aşağı ölçeklendirir ve siz yalnızca kullandığınız kadar ödersiniz.

Ortak Tarama veri kümesinin işlenmesi genellikle tek seferlik bir işlem görevidir ve bu da onu EMR Sunucusuz iş yükleri için uygun kılar.

EMR Sunucusuz uygulama oluşturma

EMR Studio konsolunda bir EMR Sunucusuz uygulaması oluşturabilirsiniz. Aşağıdaki adımları tamamlayın:

1. EMR Studio konsolunda, Uygulamalar altında Serverless Gezinti bölmesinde.
2. Klinik Uygulama oluştur.

3. Uygulama için bir ad girin ve bir Amazon EMR sürümü seçin.

4. VPC kaynaklarına erişim gerekiyorsa özelleştirilmiş bir ağ ayarı ekleyin.

5. Klinik Uygulama oluştur.

Spark sunucusuz ortamınız daha sonra hazır olacaktır.

EMR Spark Serverless'a bir iş göndermeden önce yine de bir yürütme rolü oluşturmanız gerekir. Bakınız Amazon EMR Sunucusuz kullanmaya başlarken daha fazla ayrıntı için.

Ortak Tarama verilerini EMR Serverless ile işleyin

EMR Spark Serverless uygulamanız hazır olduktan sonra verileri işlemek için aşağıdaki adımları tamamlayın:

1. Bir Conda ortamı hazırlayın ve bunu EMR Spark Serverless'ta ortam olarak kullanılacak Amazon S3'e yükleyin.
2. Çalıştırılacak komut dosyalarını bir S3 klasörüne yükleyin. Aşağıdaki örnekte iki komut dosyası vardır:
   1. imbd_extractor.py – Veri kümesinden içerik çıkarmak için özelleştirilmiş mantık. İçeriği bu yazının başlarında bulabilirsiniz.
   2. cc-pyspark/sparkcc.py – Örnek PySpark çerçevesi Ortak Tarama GitHub deposu, dahil edilmesi gereken bir şey.
3. PySpark işini EMR Serverless Spark'a gönderin. Bu örneği ortamınızda çalıştırmak için aşağıdaki parametreleri tanımlayın:
   1. Uygulama Kimliği – EMR Sunucusuz uygulamanızın uygulama kimliği.
   2. yürütme rolü-arn – EMR Sunucusuz yürütme rolünüz. Bunu oluşturmak için bkz. İş çalışma zamanı rolü oluşturma.
   3. WARC dosya konumu – WARC dosyalarınızın konumu. s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt Bu yazının başlarında edindiğiniz filtrelenmiş WARC dosya listesini içerir.
   4. spark.sql.warehouse.dir – Varsayılan depo konumu (S3 dizininizi kullanın).
   5. spark.archives – Hazırlanan Conda ortamının S3 konumu.
   6. spark.submit.pyDosyaları – Hazırlanan PySpark scripti sparkcc.py.

Aşağıdaki koda bakın:

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

İş tamamlandıktan sonra çıkarılan incelemeler Amazon S3'te depolanır. İçerikleri kontrol etmek için aşağıdaki ekran görüntüsünde gösterildiği gibi Amazon S3 Select'i kullanabilirsiniz.

Hususlar

Özelleştirilmiş kodla büyük miktarda veriyle uğraşırken dikkate alınması gereken noktalar şunlardır:

• Bazı üçüncü taraf Python kitaplıkları Conda'da mevcut olmayabilir. Bu gibi durumlarda PySpark çalışma zamanı ortamını oluşturmak için Python sanal ortamına geçebilirsiniz.
• İşlenecek çok büyük miktarda veri varsa bunu paralelleştirmek için birden fazla EMR Serverless Spark uygulaması oluşturup kullanmayı deneyin. Her uygulama, dosya listelerinin bir alt kümesiyle ilgilenir.
• Ortak Tarama verilerini filtrelerken veya işlerken Amazon S3'te yavaşlama sorunuyla karşılaşabilirsiniz. Bunun nedeni, verileri depolayan S3 kümesinin herkese açık olması ve diğer kullanıcıların da verilere aynı anda erişebilmesidir. Bu sorunu azaltmak için bir yeniden deneme mekanizması ekleyebilir veya Common Crawl S3 klasöründeki belirli verileri kendi grubunuzla senkronize edebilirsiniz.

SageMaker ile Llama 2'ye ince ayar yapın

Veriler hazırlandıktan sonra onunla bir Llama 2 modeline ince ayar yapabilirsiniz. Bunu SageMaker JumpStart'ı kullanarak herhangi bir kod yazmadan yapabilirsiniz. Daha fazla bilgi için bkz. Amazon SageMaker JumpStart'ta metin oluşturmak için Llama 2'ye ince ayar yapın.

Bu senaryoda, etki alanı uyarlama ince ayarı gerçekleştirirsiniz. Bu veri kümesinde girdi CSV, JSON veya TXT dosyasından oluşur. Tüm inceleme verilerini bir TXT dosyasına koymanız gerekir. Bunu yapmak için EMR Spark Serverless'a basit bir Spark işi gönderebilirsiniz. Aşağıdaki örnek kod pasajına bakın:

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

Eğitim verilerini hazırladıktan sonra veri konumunu girin. Eğitim veri seti, Daha sonra seçmek Tren.

Eğitim işi durumunu takip edebilirsiniz.

İnce ayarlı modeli değerlendirin

Eğitim tamamlandıktan sonra seçin Sürüş İnce ayarlı modelinizi dağıtmak için SageMaker JumpStart'ta.

Model başarıyla dağıtıldıktan sonra şunu seçin: Not Defterini Açsizi Python kodunuzu çalıştırabileceğiniz hazırlanmış bir Jupyter not defterine yönlendirir.

Dizüstü bilgisayar için Image Data Science 2.0 ve Python 3 çekirdeğini kullanabilirsiniz.

Daha sonra bu defterdeki ince ayarlı modeli ve orijinal modeli değerlendirebilirsiniz.

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)

Aşağıda aynı soru için orijinal model ve ince ayarlı model tarafından döndürülen iki yanıt bulunmaktadır.

Her iki modele de aynı cümleyi verdik: “'A Woman of Paris: A Drama of Fate' filminin eleştirisi” ve cümleyi tamamlamalarını sağladık.

Orijinal model anlamsız cümleler çıkarır:

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

..."

Buna karşılık, ince ayarlı modelin çıktıları daha çok bir film incelemesine benzer:

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

Açıkçası, ince ayarlı model bu özel senaryoda daha iyi performans gösteriyor.

Temizlemek

Bu alıştırmayı tamamladıktan sonra kaynaklarınızı temizlemek için aşağıdaki adımları tamamlayın:

1. S3 klasörünü silin temizlenmiş veri kümesini saklayan.
2. EMR Sunucusuz ortamını durdurun.
3. SageMaker uç noktasını silin LLM modelini barındıran.
4. SageMaker alanını silin not defterlerinizi çalıştırır.

Oluşturduğunuz uygulama varsayılan olarak 15 dakika işlem yapılmadığında otomatik olarak durmalıdır.

Genel olarak Athena ortamını temizlemenize gerek yoktur çünkü kullanmadığınız zamanlarda herhangi bir ücret alınmaz.

Sonuç

Bu yazıda Common Crawl veri kümesini ve LLM ince ayarı için verileri işlemek amacıyla EMR Serverless'ın nasıl kullanılacağını tanıttık. Daha sonra LLM'ye ince ayar yapmak ve herhangi bir kod olmadan dağıtmak için SageMaker JumpStart'ın nasıl kullanılacağını gösterdik. EMR Serverless'ın daha fazla kullanım örneği için bkz. Amazon EMR Sunucusuz. Amazon SageMaker JumpStart'ta modelleri barındırma ve ince ayar yapma hakkında daha fazla bilgi için bkz. Sagemaker JumpStart belgeleri.

Yazarlar Hakkında

Shijian Tang Amazon Web Services'te Analitik Uzmanı Çözüm Mimarıdır.

Matthew Liem Amazon Web Services'te Kıdemli Çözüm Mimarisi Yöneticisidir.

Dalei Xu Amazon Web Services'te Analitik Uzmanı Çözüm Mimarıdır.

Yuanjun Xiao Amazon Web Services'te Kıdemli Çözüm Mimarıdır.