$  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

توضح عبارات SQL السابقة كيفية إنشاء جدول Athena وإضافة أقسام وتشغيل الاستعلام.

تصفية البيانات من مجموعة بيانات الزحف المشترك

كما ترون من جملة إنشاء جدول SQL، هناك العديد من الحقول التي يمكن أن تساعد في تصفية البيانات. على سبيل المثال، إذا كنت تريد الحصول على عدد المستندات الصينية خلال فترة محددة، فيمكن أن تكون عبارة SQL كما يلي:

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

إذا كنت تريد إجراء مزيد من المعالجة، فيمكنك حفظ النتائج في حاوية S3 أخرى.

تحليل البيانات التي تمت تصفيتها

• مستودع الزحف المشترك على GitHub يوفر العديد من أمثلة PySpark لمعالجة البيانات الأولية.

دعونا نلقي نظرة على مثال للتشغيل server_count.py (مثال على البرنامج النصي المقدم من Common Crawl GitHub repo) على البيانات الموجودة في s3://commoncrawl/crawl-data/CC-MAIN-2023-23/segments/1685224643388.45/warc/.

أولاً، أنت بحاجة إلى بيئة Spark، مثل EMR Spark. على سبيل المثال، يمكنك تشغيل Amazon EMR على مجموعة EC2 في us-east-1 (لأن مجموعة البيانات موجودة us-east-1). يمكن أن يساعدك استخدام EMR في مجموعة EC2 في إجراء الاختبارات قبل إرسال المهام إلى بيئة الإنتاج.

بعد تشغيل EMR على مجموعة EC2، يتعين عليك تسجيل الدخول عبر SSH إلى العقدة الأساسية للمجموعة. ثم قم بحزم بيئة Python وأرسل البرنامج النصي (راجع ملف وثائق كوندا لتثبيت مينيكوندا):

#  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. لأغراض العرض التوضيحي، يمكنك تحسين وقت المعالجة باستخدام الاستراتيجيات التالية:

• تحميل الملف s3://commoncrawl/crawl-data/CC-MAIN-2023-23/warc.paths.gz إلى جهازك المحلي، وقم بفك ضغطه، ثم قم بتحميله على HDFS أو Amazon S3. وذلك لأن الملف .gzip غير قابل للتقسيم. تحتاج إلى فك ضغطه لمعالجة هذا الملف بالتوازي.
• تعديل warc.path الملف، واحذف معظم أسطره، واحتفظ بسطرين فقط لتسيير المهمة بشكل أسرع.

بعد الانتهاء من المهمة، يمكنك رؤية النتيجة s3://xxxx-common-crawl/output/، على شكل باركيه.

تنفيذ منطق امتلاك مخصص

يوفر Common Crawl GitHub repo طريقة شائعة لمعالجة ملفات WARC. بشكل عام، يمكنك تمديد CCSparkJob لتجاوز طريقة واحدة (process_record)، وهو كافٍ لكثير من الحالات.

دعونا نلقي نظرة على مثال للحصول على تقييمات IMDB للأفلام الحديثة. أولاً، عليك تصفية الملفات الموجودة على موقع 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

ثم يمكنك الحصول على قوائم ملفات WARC التي تحتوي على بيانات مراجعة IMDB، وحفظ أسماء ملفات WARC كقائمة في ملف نصي.

وبدلاً من ذلك، يمكنك استخدام EMR Spark للحصول على قائمة ملفات WARC وتخزينها في Amazon S3. على سبيل المثال:

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

يجب أن يبدو ملف الإخراج مشابهًا لـ s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt.

الخطوة التالية هي استخراج مراجعات المستخدم من ملفات WARC هذه. يمكنك تمديد CCSparkJob لتجاوز process_record() الأسلوب:

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

يمكنك حفظ البرنامج النصي السابق باسم imdb_extractor.py، والذي ستستخدمه في الخطوات التالية. بعد إعداد البيانات والبرامج النصية، يمكنك استخدام EMR Serverless لمعالجة البيانات التي تمت تصفيتها.

EMR بدون خادم

EMR Serverless هو خيار نشر بدون خادم لتشغيل تطبيقات تحليلات البيانات الضخمة باستخدام أطر عمل مفتوحة المصدر مثل Apache Spark وHive دون تكوين مجموعات أو خوادم وإدارتها وتوسيع نطاقها.

باستخدام EMR Serverless، يمكنك تشغيل أحمال عمل التحليلات على أي نطاق من خلال القياس التلقائي الذي يغير حجم الموارد في ثوانٍ لتلبية أحجام البيانات المتغيرة ومتطلبات المعالجة. يقوم EMR Serverless تلقائيًا بتوسيع نطاق الموارد لأعلى ولأسفل لتوفير المقدار المناسب من السعة لتطبيقك، ولا تدفع إلا مقابل ما تستخدمه.

تعد معالجة مجموعة بيانات الزحف المشترك بشكل عام مهمة معالجة لمرة واحدة، مما يجعلها مناسبة لأحمال عمل EMR بدون خادم.

إنشاء تطبيق EMR بدون خادم

يمكنك إنشاء تطبيق EMR Serverless على وحدة تحكم EMR Studio. أكمل الخطوات التالية:

1. في وحدة تحكم استوديو EMR ، اختر التطبيقات مع Serverless في جزء التنقل.
2. اختار إنشاء التطبيق.

3. أدخل اسمًا للتطبيق واختر إصدار Amazon EMR.

4. إذا كان الوصول إلى موارد VPC مطلوبًا، فقم بإضافة إعداد شبكة مخصص.

5. اختار إنشاء التطبيق.

ستكون بيئة Spark بدون خادم جاهزة بعد ذلك.

قبل أن تتمكن من إرسال مهمة إلى EMR Spark Serverless، لا تزال بحاجة إلى إنشاء دور تنفيذ. تشير إلى الشروع في العمل مع Amazon EMR Serverless لمزيد من التفاصيل.

معالجة بيانات الزحف الشائعة باستخدام EMR Serverless

بعد أن يصبح تطبيق EMR Spark Serverless جاهزًا، أكمل الخطوات التالية لمعالجة البيانات:

1. قم بإعداد بيئة Conda وقم بتحميلها إلى Amazon S3، والتي سيتم استخدامها كبيئة في EMR Spark Serverless.
2. قم بتحميل البرامج النصية ليتم تشغيلها إلى مجموعة S3. في المثال التالي، هناك نوعان من البرامج النصية:
   1. imbd_extractor.py – منطق مخصص لاستخراج المحتويات من مجموعة البيانات. محتويات يمكن العثور عليها في وقت سابق من هذا المنصب.
   2. cc-pyspark/sparkcc.py - مثال إطار عمل PySpark من ريبو الزحف المشترك على جيثب، وهو أمر ضروري لإدراجه.
3. أرسل مهمة PySpark إلى EMR Serverless Spark. حدد المعلمات التالية لتشغيل هذا المثال في بيئتك:
   1. رقم الاستمارة – معرف التطبيق لتطبيق EMR Serverless الخاص بك.
   2. تنفيذ الدور-arn – دور التنفيذ بدون خادم EMR الخاص بك. لإنشائه، راجع إنشاء دور وقت التشغيل الوظيفي.
   3. موقع ملف WARC - موقع ملفات WARC الخاصة بك. s3://xxxx-common-crawl/warclist/imdb_warclist/part-00000-6af12797-0cdc-4ef2-a438-cf2b935f2ffd-c000.txt يحتوي على قائمة ملفات WARC التي تمت تصفيتها، والتي حصلت عليها سابقًا في هذا المنشور.
   4. spark.sql.warehouse.dir – موقع المستودع الافتراضي (استخدم دليل S3 الخاص بك).
   5. spark.archives – موقع S3 لبيئة كوندا المعدة.
   6. spark.submit.pyFiles - البرنامج النصي PySpark المُجهز sparkcc.py.

انظر الكود التالي:

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

بعد اكتمال المهمة، يتم تخزين المراجعات المستخرجة في Amazon S3. للتحقق من المحتويات، يمكنك استخدام Amazon S3 Select، كما هو موضح في لقطة الشاشة التالية.

الاعتبارات

فيما يلي النقاط التي يجب مراعاتها عند التعامل مع كميات هائلة من البيانات باستخدام تعليمات برمجية مخصصة:

• قد لا تتوفر بعض مكتبات Python التابعة لجهات خارجية في Conda. في مثل هذه الحالات، يمكنك التبديل إلى بيئة Python الافتراضية لإنشاء بيئة تشغيل PySpark.
• إذا كانت هناك كمية هائلة من البيانات المراد معالجتها، فحاول إنشاء واستخدام العديد من تطبيقات EMR Serverless Spark لموازاة ذلك. يتعامل كل تطبيق مع مجموعة فرعية من قوائم الملفات.
• قد تواجه مشكلة تباطؤ في Amazon S3 عند تصفية بيانات الزحف المشترك أو معالجتها. وذلك لأن حاوية S3 التي تخزن البيانات يمكن الوصول إليها بشكل عام، ويمكن للمستخدمين الآخرين الوصول إلى البيانات في نفس الوقت. للتخفيف من هذه المشكلة، يمكنك إضافة آلية إعادة المحاولة أو مزامنة بيانات معينة من حاوية Common Crawl S3 إلى الحاوية الخاصة بك.

قم بضبط Llama 2 باستخدام SageMaker

بعد إعداد البيانات، يمكنك ضبط نموذج Llama 2 باستخدامها. يمكنك القيام بذلك باستخدام SageMaker JumpStart، دون كتابة أي تعليمات برمجية. لمزيد من المعلومات، راجع قم بضبط Llama 2 لإنشاء النص على Amazon SageMaker JumpStart.

في هذا السيناريو، يمكنك إجراء ضبط دقيق للتكيف مع المجال. باستخدام مجموعة البيانات هذه، يتكون الإدخال من ملف CSV أو JSON أو TXT. تحتاج إلى وضع جميع بيانات المراجعة في ملف TXT. للقيام بذلك، يمكنك إرسال مهمة Spark مباشرة إلى EMR Spark Serverless. راجع مقتطف التعليمات البرمجية النموذجي التالي:

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

بعد الانتهاء من تجهيز بيانات التدريب، قم بإدخال موقع البيانات الخاص به مجموعة بيانات التدريب، ثم اختر قطار.

يمكنك تتبع حالة مهمة التدريب.

تقييم النموذج المضبوط

بعد الانتهاء من التدريب، اختر نشر في SageMaker JumpStart لنشر نموذجك المضبوط بدقة.

بعد نشر النموذج بنجاح، اختر افتح المفكرة، والذي يعيد توجيهك إلى دفتر Jupyter المُجهز حيث يمكنك تشغيل كود Python الخاص بك.

يمكنك استخدام الصورة Data Science 2.0 وPython 3 kernel لدفتر الملاحظات.

بعد ذلك، يمكنك تقييم النموذج المضبوط والنموذج الأصلي في هذا الكمبيوتر الدفتري.

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 Woman of Paris: A Drama of Fate هي" ودعناهما يكملان الجملة.

يُخرج النموذج الأصلي جملًا لا معنى لها:

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

..."

في المقابل، فإن مخرجات النموذج المضبوط تشبه إلى حد كبير مراجعة الفيلم:

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

من الواضح أن النموذج المضبوط يعمل بشكل أفضل في هذا السيناريو المحدد.

تنظيف

بعد الانتهاء من هذا التمرين، أكمل الخطوات التالية لتنظيف مواردك:

1. احذف حاوية S3 الذي يخزن مجموعة البيانات التي تم تنظيفها.
2. أوقف بيئة EMR بدون خادم.
3. احذف نقطة نهاية SageMaker التي تستضيف نموذج LLM.
4. احذف مجال SageMaker الذي يدير دفاتر الملاحظات الخاصة بك.

يجب أن يتوقف التطبيق الذي قمت بإنشائه تلقائيًا بعد 15 دقيقة من عدم النشاط بشكل افتراضي.

بشكل عام، لا تحتاج إلى تنظيف بيئة Athena لأنه لا توجد رسوم عند عدم استخدامها.

وفي الختام

في هذا المنشور، قدمنا ​​مجموعة بيانات Common Crawl وكيفية استخدام EMR Serverless لمعالجة البيانات من أجل الضبط الدقيق لـ LLM. ثم أوضحنا كيفية استخدام SageMaker JumpStart لضبط LLM ونشره بدون أي تعليمات برمجية. لمزيد من حالات استخدام EMR Serverless، راجع أمازون EMR بدون خادم. لمزيد من المعلومات حول استضافة النماذج وضبطها على Amazon SageMaker JumpStart، راجع وثائق Sagemaker JumpStart.

حول المؤلف

شيجيان تانغ هو مهندس حلول متخصص في التحليلات في Amazon Web Services.

ماثيو ليم هو مدير أول لهندسة الحلول في Amazon Web Services.

دالي شو هو مهندس حلول متخصص في التحليلات في Amazon Web Services.

يوان جون شياو هو مهندس حلول أول في Amazon Web Services.