$  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

Các câu lệnh SQL trước đây minh họa cách tạo bảng Athena, thêm phân vùng và chạy truy vấn.

Lọc dữ liệu từ tập dữ liệu Thu thập thông tin chung

Như bạn có thể thấy từ câu lệnh SQL tạo bảng, có một số trường có thể giúp lọc dữ liệu. Ví dụ: nếu bạn muốn lấy số lượng tài liệu tiếng Trung trong một khoảng thời gian cụ thể thì câu lệnh SQL có thể như sau:

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

Nếu muốn xử lý thêm, bạn có thể lưu kết quả vào một bộ chứa S3 khác.

Phân tích dữ liệu đã lọc

Sản phẩm Kho lưu trữ GitHub thu thập thông tin chung cung cấp một số ví dụ PySpark để xử lý dữ liệu thô.

Hãy xem một ví dụ về việc chạy server_count.py (tập lệnh mẫu được cung cấp bởi kho lưu trữ Common Crawl GitHub) trên dữ liệu nằm trong s3://commoncrawl/crawl-data/CC-MAIN-2023-23/segments/1685224643388.45/warc/.

Đầu tiên, bạn cần có môi trường Spark, chẳng hạn như EMR Spark. Ví dụ: bạn có thể khởi chạy Amazon EMR trên cụm EC2 trong us-east-1 (vì tập dữ liệu nằm trong us-east-1). Sử dụng EMR trên cụm EC2 có thể giúp bạn thực hiện các thử nghiệm trước khi gửi công việc đến môi trường sản xuất.

Sau khi khởi chạy EMR trên cụm EC2, bạn cần thực hiện đăng nhập SSH vào nút chính của cụm. Sau đó, đóng gói môi trường Python và gửi tập lệnh (tham khảo phần Tài liệu Conda để cài đặt 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

Có thể mất thời gian để xử lý tất cả các tham chiếu trong warc.path. Đối với mục đích demo, bạn có thể cải thiện thời gian xử lý bằng các chiến lược sau:

• Tải xuống tệp tin s3://commoncrawl/crawl-data/CC-MAIN-2023-23/warc.paths.gz vào máy cục bộ của bạn, giải nén nó rồi tải nó lên HDFS hoặc Amazon S3. Điều này là do tệp .gzip không thể chia tách được. Bạn cần giải nén nó để xử lý file này song song.
• Sửa đổi warc.path tập tin, xóa hầu hết các dòng của nó và chỉ giữ lại hai dòng để công việc chạy nhanh hơn nhiều.

Sau khi hoàn thành công việc, bạn có thể xem kết quả trong s3://xxxx-common-crawl/output/, ở định dạng Parquet.

Triển khai logic sở hữu tùy chỉnh

Kho lưu trữ Common Crawl GitHub cung cấp một cách tiếp cận chung để xử lý các tệp WARC. Nói chung, bạn có thể mở rộng CCSparkJob để ghi đè một phương thức duy nhất (process_record), điều này là đủ cho nhiều trường hợp.

Hãy xem một ví dụ để lấy các đánh giá IMDB về các bộ phim gần đây. Trước tiên, bạn cần lọc các tệp trên trang 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

Sau đó, bạn có thể lấy danh sách tệp WARC chứa dữ liệu đánh giá IMDB và lưu tên tệp WARC dưới dạng danh sách trong tệp văn bản.

Ngoài ra, bạn có thể sử dụng EMR Spark lấy danh sách tệp WARC và lưu trữ nó trong Amazon S3. Ví dụ:

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

Tệp đầu ra sẽ trông giống như s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt.

Bước tiếp theo là trích xuất đánh giá của người dùng từ các tệp WARC này. Bạn có thể mở rộng CCSparkJob ghi đè lên process_record() phương pháp:

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

Bạn có thể lưu tập lệnh trước dưới dạng imdb_extractor.py mà bạn sẽ sử dụng trong các bước sau. Sau khi đã chuẩn bị xong dữ liệu và tập lệnh, bạn có thể sử dụng EMR Serverless để xử lý dữ liệu đã được lọc.

EMR không có máy chủ

EMR Serverless là một tùy chọn triển khai không có máy chủ để chạy các ứng dụng phân tích dữ liệu lớn bằng cách sử dụng các khung nguồn mở như Apache Spark và Hive mà không cần định cấu hình, quản lý và mở rộng quy mô cụm hoặc máy chủ.

Với EMR Serverless, bạn có thể chạy khối lượng công việc phân tích ở mọi quy mô với tính năng tự động thay đổi quy mô giúp thay đổi kích thước tài nguyên trong vài giây để đáp ứng các yêu cầu xử lý và khối lượng dữ liệu luôn thay đổi. EMR Serverless tự động tăng giảm quy mô tài nguyên để cung cấp dung lượng phù hợp cho ứng dụng của bạn và bạn chỉ trả tiền cho những gì bạn sử dụng.

Việc xử lý tập dữ liệu Thu thập thông tin chung thường là tác vụ xử lý một lần, khiến nó phù hợp với khối lượng công việc EMR Serverless.

Tạo ứng dụng EMR Serverless

Bạn có thể tạo ứng dụng EMR Serverless trên bảng điều khiển EMR Studio. Hoàn thành các bước sau:

1. Trên bảng điều khiển EMR Studio, chọn Ứng dụng Dưới Không có máy chủ trong khung điều hướng.
2. Chọn Tạo ứng dụng.

3. Cung cấp tên cho ứng dụng và chọn phiên bản Amazon EMR.

4. Nếu cần có quyền truy cập vào tài nguyên VPC, hãy thêm cài đặt mạng tùy chỉnh.

5. Chọn Tạo ứng dụng.

Môi trường không có máy chủ Spark của bạn sau đó sẽ sẵn sàng.

Trước khi có thể gửi công việc tới EMR Spark Serverless, bạn vẫn cần tạo vai trò thực thi. tham khảo Bắt đầu với Amazon EMR Serverless để biết thêm chi tiết.

Xử lý dữ liệu thu thập thông tin phổ biến với EMR Serverless

Sau khi ứng dụng EMR Spark Serverless của bạn sẵn sàng, hãy hoàn thành các bước sau để xử lý dữ liệu:

1. Chuẩn bị môi trường Conda và tải nó lên Amazon S3. Môi trường này sẽ được sử dụng làm môi trường trong EMR Spark Serverless.
2. Tải tập lệnh lên để chạy vào vùng lưu trữ S3. Trong ví dụ sau, có hai tập lệnh:
   1. imbd_extractor.py – Logic tùy chỉnh để trích xuất nội dung từ tập dữ liệu. Nội dung có thể được tìm thấy trước đó trong bài viết này.
   2. cc-pyspark/sparkcc.py – Ví dụ khung PySpark từ Kho GitHub thu thập thông tin chung, điều cần thiết phải được đưa vào.
3. Gửi công việc PySpark tới EMR Serverless Spark. Xác định các tham số sau để chạy ví dụ này trong môi trường của bạn:
   1. id ứng dụng – ID ứng dụng của ứng dụng EMR Serverless của bạn.
   2. vai trò thực thi-arn – Vai trò thực thi EMR Serverless của bạn. Để tạo nó, hãy tham khảo Tạo vai trò thời gian chạy công việc.
   3. Vị trí tệp WARC – Vị trí tệp WARC của bạn. s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt chứa danh sách tệp WARC đã lọc mà bạn đã lấy được trước đó trong bài đăng này.
   4. spark.sql.warehouse.dir – Vị trí kho mặc định (sử dụng thư mục S3 của bạn).
   5. spark.archives – Vị trí S3 của môi trường Conda đã chuẩn bị.
   6. spark.submit.pyFiles – Tập lệnh PySpark đã được chuẩn bị sẵn sparkcc.py.

Xem mã sau đây:

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

Sau khi công việc hoàn tất, các đánh giá được trích xuất sẽ được lưu trữ trong Amazon S3. Để kiểm tra nội dung, bạn có thể sử dụng Amazon S3 Select, như trong ảnh chụp màn hình sau.

Những cân nhắc

Sau đây là những điểm cần cân nhắc khi xử lý lượng dữ liệu khổng lồ bằng mã tùy chỉnh:

• Một số thư viện Python của bên thứ ba có thể không có sẵn trong Conda. Trong những trường hợp như vậy, bạn có thể chuyển sang môi trường ảo Python để xây dựng môi trường thời gian chạy PySpark.
• Nếu có một lượng lớn dữ liệu cần xử lý, hãy thử tạo và sử dụng nhiều ứng dụng EMR Serverless Spark để song song hóa nó. Mỗi ứng dụng xử lý một tập hợp con các danh sách tập tin.
• Bạn có thể gặp phải sự cố làm chậm với Amazon S3 khi lọc hoặc xử lý dữ liệu Thu thập thông tin chung. Điều này là do bộ chứa S3 lưu trữ dữ liệu có thể truy cập công khai và những người dùng khác có thể truy cập dữ liệu cùng lúc. Để giảm thiểu sự cố này, bạn có thể thêm cơ chế thử lại hoặc đồng bộ hóa dữ liệu cụ thể từ vùng lưu trữ Common Crawl S3 sang vùng lưu trữ của riêng bạn.

Tinh chỉnh Llama 2 bằng SageMaker

Sau khi dữ liệu được chuẩn bị, bạn có thể tinh chỉnh mô hình Llama 2 bằng dữ liệu đó. Bạn có thể làm như vậy bằng SageMaker JumpStart mà không cần viết bất kỳ mã nào. Để biết thêm thông tin, hãy tham khảo Tinh chỉnh Llama 2 để tạo văn bản trên Amazon SageMaker JumpStart.

Trong trường hợp này, bạn thực hiện tinh chỉnh thích ứng miền. Với tập dữ liệu này, đầu vào bao gồm tệp CSV, JSON hoặc TXT. Bạn cần đặt tất cả dữ liệu đánh giá vào một tệp TXT. Để làm như vậy, bạn có thể gửi công việc Spark đơn giản tới EMR Spark Serverless. Xem đoạn mã mẫu sau:

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

Sau khi bạn chuẩn bị dữ liệu huấn luyện, hãy nhập vị trí dữ liệu cho Tập dữ liệu huấn luyện, sau đó chọn Train.

Bạn có thể theo dõi tình trạng công việc đào tạo.

Đánh giá mô hình hiệu chỉnh

Sau khi đào tạo xong, chọn Triển khai trong SageMaker JumpStart để triển khai mô hình tinh chỉnh của bạn.

Sau khi mô hình được triển khai thành công, chọn Mở Notebook, chuyển hướng bạn đến sổ ghi chép Jupyter đã chuẩn bị sẵn nơi bạn có thể chạy mã Python của mình.

Bạn có thể sử dụng hình ảnh Data Science 2.0 và hạt nhân Python 3 cho sổ ghi chép.

Sau đó, bạn có thể đánh giá mô hình đã tinh chỉnh và mô hình gốc trong sổ ghi chép này.

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)

Sau đây là hai câu trả lời do mô hình ban đầu và mô hình tinh chỉnh trả về cho cùng một câu hỏi.

Chúng tôi cung cấp cho cả hai người mẫu một câu giống nhau: “Bài đánh giá về phim 'A Woman of Paris: A Drama of Fate' là” và để họ hoàn thành câu.

Mô hình ban đầu đưa ra những câu vô nghĩa:

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

..."

Ngược lại, kết quả đầu ra của mô hình tinh chỉnh giống như một bản đánh giá phim hơn:

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

Rõ ràng, mô hình được tinh chỉnh sẽ hoạt động tốt hơn trong trường hợp cụ thể này.

Làm sạch

Sau khi bạn hoàn thành bài tập này, hãy hoàn thành các bước sau để dọn dẹp tài nguyên của bạn:

1. Xóa bộ chứa S3 lưu trữ tập dữ liệu đã được làm sạch.
2. Dừng môi trường EMR Serverless.
3. Xóa điểm cuối SageMaker lưu trữ mô hình LLM.
4. Xóa miền SageMaker chạy sổ ghi chép của bạn.

Theo mặc định, ứng dụng bạn đã tạo sẽ tự động dừng sau 15 phút không hoạt động.

Nói chung, bạn không cần phải dọn dẹp môi trường Athena vì bạn không phải trả phí khi không sử dụng.

Kết luận

Trong bài đăng này, chúng tôi đã giới thiệu bộ dữ liệu Thu thập thông tin chung và cách sử dụng EMR Serverless để xử lý dữ liệu nhằm tinh chỉnh LLM. Sau đó, chúng tôi đã trình bày cách sử dụng SageMaker JumpStart để tinh chỉnh LLM và triển khai nó mà không cần bất kỳ mã nào. Để biết thêm các trường hợp sử dụng EMR Serverless, hãy tham khảo Amazon EMR không có máy chủ. Để biết thêm thông tin về mô hình lưu trữ và tinh chỉnh trên Amazon SageMaker JumpStart, hãy tham khảo Tài liệu Sagemaker JumpStart.

Về các tác giả

Đường Thạch Kiến là Kiến trúc sư giải pháp chuyên gia phân tích tại Amazon Web Services.

Matthêu Liêm là Giám đốc kiến ​​trúc giải pháp cấp cao tại Amazon Web Services.

Dalei Xu là Kiến trúc sư giải pháp chuyên gia phân tích tại Amazon Web Services.

Nguyên Quân Tiêu là Kiến trúc sư giải pháp cấp cao tại Amazon Web Services.