PySpark لمعالجة اللغات الطبيعية على Dataproc

1. نظرة عامة

معالجة اللغات الطبيعية (NLP) هي دراسة لاستخلاص الأفكار وإجراء تحليلات على البيانات النصية. مع استمرار تزايد مقدار المحتوى المكتوب على الإنترنت، أصبحت الآن أكثر من أي وقت مضى تسعى المؤسسات إلى الاستفادة من النصوص لاكتساب معلومات ذات صلة بأعمالها.

يمكن استخدام "تعلُّم الآلة" (NLP) في تنفيذ كل المهام، بدءًا من ترجمة اللغات إلى تحليل الشعور ووصولاً إلى تكوين جمل من الصفر وغير ذلك الكثير. وهو مجال بحث نشط يغير طريقة عملنا مع النصوص.

وسنستكشف كيفية استخدام معالجة اللغات الطبيعية على كميات كبيرة من البيانات النصية على نطاق واسع. يمكن أن تكون هذه بالتأكيد مهمة شاقة! ولحسن الحظ، سنستفيد من مكتبات مثل Spark MLlib وspark-nlp لتسهيل هذه المهمة.

2. حالة الاستخدام

كبير علماء البيانات في مؤسستنا (الخيالية) "FoodCorp" مهتمًا بمعرفة المزيد عن الاتجاهات في صناعة الأغذية. يمكننا الوصول إلى مجموعة من البيانات النصية على شكل مشاركات من Reddit subreddit r/food لاستكشاف ما يتحدث عنه الأشخاص.

ويتمثل أحد المناهج للقيام بذلك في طريقة معالجة اللغات الطبيعية المعروفة باسم "نمذجة الموضوع". نمذجة الموضوعات هي طريقة إحصائية يمكنها تحديد الاتجاهات في المعاني الدلالية لمجموعة من المستندات. أو بعبارةٍ أخرى، يمكننا إنشاء نموذج موضوع على مجموعة "مشاركات" Reddit التي تنشئ قائمة بـ "المواضيع" أو مجموعات من الكلمات التي تصف الاتجاه

لوضع نموذجنا، سنستخدم خوارزمية تُسمى Latent Dirichlet Allocation (LDA) والتي غالبًا ما تُستخدم لتجميع النص. يمكنك العثور على مقدمة رائعة حول LDA هنا.

3- إنشاء مشروع

إذا لم يكن لديك حساب Google (Gmail أو Google Apps)، يجب عليك إنشاء حساب. سجِّل الدخول إلى وحدة تحكُّم Google Cloud Platform ( console.cloud.google.com) وأنشئ مشروعًا جديدًا:

بعد ذلك، عليك تفعيل الفوترة في Cloud Console لاستخدام موارد Google Cloud.

لا ينبغي أن يكلفك هذا الدرس التطبيقي أكثر من بضعة دولارات، ولكن قد تزيد التكاليف إذا قررت استخدام المزيد من الموارد أو إذا تركتها قيد التشغيل. يشرح كل درس تطبيقي حول الترميز PySpark-BigQuery وSpark-NLP تعليمات "تنظيف البيانات". في النهاية.

المستخدمون الجدد في Google Cloud Platform مؤهّلون للحصول على فترة تجريبية مجانية بقيمة 300 دولار أمريكي.

4. إعداد البيئة

نحتاج أولاً إلى تفعيل Dataproc وواجهات Compute Engine API.

انقر على رمز القائمة في أعلى يمين الشاشة.

اختر مدير واجهة برمجة التطبيقات من القائمة المنسدلة.

انقر على تفعيل واجهات برمجة التطبيقات والخدمات (Enable APIs and Services).

ابحث عن "Compute Engine" في مربع البحث. انقر على "Google Compute Engine API". في قائمة النتائج التي تظهر.

في صفحة Google Compute Engine، انقر على تفعيل.

وبعد تفعيلها، انقر على السهم الذي يشير إلى اليمين للرجوع.

ابحث الآن عن "Google Dataproc API". وتمكينه أيضًا.

بعد ذلك، افتح Cloud Shell بالنقر على الزر في الزاوية العلوية اليسرى من وحدة التحكم في السحابة الإلكترونية:

سنضع بعض متغيرات البيئة التي يمكننا الرجوع إليها أثناء متابعة التمرين المعملي حول الترميز. أولاً، اختَر اسمًا لمجموعة Dataproc التي سننشئها، مثل "my-cluster"، واضبطها في بيئتك. يمكنك استخدام أي اسم تريده.

CLUSTER_NAME=my-cluster

بعد ذلك، اختَر منطقة من إحدى المناطق المتاحة هنا. على سبيل المثال: us-east1-b.

REGION=us-east1

وأخيرًا، نحتاج إلى تعيين حزمة المصدر التي ستقرأ مهمتنا البيانات منها. تتوفّر لدينا عيّنة من البيانات في الحزمة bm_reddit، ولكن يمكنك استخدام البيانات التي أنشأتها من خلال PySpark لأغراض المعالجة المسبقة لبيانات BigQuery إذا كنت قد أكملت هذه البيانات قبل هذه الحزمة.

BUCKET_NAME=bm_reddit

بعد ضبط متغيّرات البيئة، سننفِّذ الأمر التالي لإنشاء مجموعة Dataproc cluster:

 gcloud beta dataproc clusters create ${CLUSTER_NAME} \
     --region ${REGION} \
     --metadata 'PIP_PACKAGES=google-cloud-storage spark-nlp==2.7.2' \
     --worker-machine-type n1-standard-8 \
     --num-workers 4 \
     --image-version 1.4-debian10 \
     --initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh \
     --optional-components=JUPYTER,ANACONDA \
     --enable-component-gateway

لنطّلِع على كل من هذه الأوامر:

gcloud beta dataproc clusters create ${CLUSTER_NAME}: سيبدأ إنشاء مجموعة Dataproc بالاسم الذي قدّمته سابقًا. ونضمِّن "beta" هنا لتفعيل الميزات التجريبية في Dataproc، مثل "مدخل المكوّنات"، الذي سنناقشه أدناه.

--zone=${ZONE}: يحدد موقع المجموعة العنقودية.

--worker-machine-type n1-standard-8: هذا هو نوع الجهاز الذي يمكن استخدامه للعاملين.

--num-workers 4: سيكون هناك أربعة عاملين في مجموعتنا.

--image-version 1.4-debian9: يشير هذا إلى نسخة الصورة التي سنستخدمها من Dataproc.

--initialization-actions ...: إجراءات الإعداد هي نصوص برمجية مخصّصة يتم تنفيذها عند إنشاء المجموعات والموظفين. ويمكن إنشاء هذه المجموعات من خلال المستخدم وتخزينها في حزمة GCS أو الرجوع إليها من خلال الحزمة العلنية dataproc-initialization-actions. سيسمح إجراء الإعداد المضمّن هنا بعمليات تثبيت حزمة Python باستخدام Pip، على النحو الوارد مع علامة --metadata.

--metadata 'PIP_PACKAGES=google-cloud-storage spark-nlp': هذه قائمة بالحِزم مفصولة بمسافات للتثبيت في Dataproc. في هذه الحالة، سنقوم بتثبيت google-cloud-storage مكتبة برامج Python وspark-nlp.

--optional-components=ANACONDA: المكوّنات الاختيارية هي حِزم شائعة يتم استخدامها مع Dataproc ويتم تثبيتها تلقائيًا على مجموعات Dataproc أثناء إنشائها. من بين مزايا استخدام المكونات الاختيارية بدلاً من إجراءات الإعداد زيادة سرعة بدء التشغيل واختبار إصدارات Dataproc محددة. بشكل عام، تعد أكثر موثوقية.

--enable-component-gateway: تتيح لنا هذه العلامة الاستفادة من مدخل مكونات Dataproc لعرض واجهات المستخدم الشائعة، مثل Zeppelin أو Jupyter أو سجلّ Spark سجلّ. ملاحظة: يتطلب بعضها المكون الاختياري المرتبط.

للحصول على مقدمة أكثر تفصيلاً حول Dataproc، يُرجى الاطّلاع على هذا الدرس التطبيقي حول الترميز.

بعد ذلك، شغِّل الأوامر التالية في Cloud Shell لاستنساخ المستودع باستخدام نموذج الرمز البرمجي والقرص المضغوط في الدليل الصحيح:

cd
git clone https://github.com/GoogleCloudPlatform/cloud-dataproc
cd cloud-dataproc/codelabs/spark-nlp

5- Spark MLlib (سبارك)

Spark MLlib هي مكتبة قابلة للتوسّع في مجال تعلُّم الآلة ومكتوبة في Apache Spark. يستطيع MLlib تحليل كميات كبيرة من البيانات من خلال الاستفادة من كفاءة Spark مع حزمة من خوارزميات التعلم الآلي المحسّنة. وتضم واجهات برمجة تطبيقات في Java وScala وPython وR. سنركز في هذا الدرس التطبيقي على لغة Python.

يحتوي MLlib على مجموعة كبيرة من المحولات والمقدرات. المحوِّل هو أداة يمكنها تغيير بياناتك أو تغييرها، عادةً باستخدام دالة transform()، بينما أداة التقدير هي خوارزمية معدّة مسبقًا يمكنك تدريب بياناتك عليها، وعادةً باستخدام دالة fit().

تشمل أمثلة المحوِّلات ما يلي:

• إنشاء رمز مميّز (إنشاء متجه للأرقام من سلسلة من الكلمات)
• ترميز واحد فعال (إنشاء متجه متناثر للأرقام يمثل الكلمات المتوفرة في سلسلة)
• مزيل كلمات الإيقاف (إزالة الكلمات التي لا تضيف قيمة دلالية إلى سلسلة)

تشمل أمثلة المقدرات ما يلي:

• التصنيف (هل هذه تفاحة أم برتقال؟)
• الانحدار (كم يجب أن تُكلّف هذه التفاحة؟)
• التجميع العنقودي (ما مدى تشابه جميع مجموعات التفاح مع بعضها البعض؟)
• أشجار القرارات (إذا كان اللون == برتقاليًا، فهذا يعني أنه برتقالي. وإلا فإنها تفاحة)
• خفض الأبعاد (هل يمكننا إزالة بعض الخصائص من مجموعة البيانات والاستمرار في التفريق بين تفاحة وبرتقال؟).

يحتوي MLlib أيضًا على أدوات للطرق الشائعة الأخرى في التعلم الآلي مثل ضبط المعلمة الفائقة والتحديد بالإضافة إلى التحقق المتبادل.

بالإضافة إلى ذلك، يحتوي MLlib على واجهة برمجة تطبيقات Pipelines، التي تسمح لك بإنشاء مسارات لتحويل البيانات باستخدام محوّلات مختلفة يمكن إعادة تنفيذها.

6- Spark-NLP

Spark-nlp هي مكتبة أنشأتها مؤسسة John Snow Labs لتنفيذ مهام فعّالة لمعالجة اللغات الطبيعية باستخدام Spark. يحتوي على أدوات مدمجة تسمى تعليقات توضيحية للمهام الشائعة، مثل:

• إنشاء رمز مميّز (إنشاء متجه للأرقام من سلسلة من الكلمات)
• إنشاء تضمينات الكلمات (تحديد العلاقة بين الكلمات عبر المتجهات)
• علامات جزء من الكلام (أي الكلمات هي أسماء؟ ما هي الأفعال؟)

على الرغم من أنّ spark-nlp خارج نطاق هذا الدرس التطبيقي حول الترميز، تندمج أيضًا بشكل رائع مع TensorFlow.

وربما يعزز Spark-NLP قدرات Spark MLlib بشكلٍ كبير من خلال توفير مكوّنات يتم إدراجها بسهولة في MLlib Pipelines.

7. أفضل الممارسات لمعالجة اللغات الطبيعية

قبل أن نتمكن من استخراج المعلومات المفيدة من البيانات، نحتاج إلى إجراء بعض الإجراءات المتعلقة بتنظيم البيانات. في ما يلي خطوات المعالجة المسبقة التي سنتّخذها:

إنشاء رمز مميّز

أول شيء نريد فعله بشكل تقليدي هو "إضافة الرموز المميزة" البيانات. ويشمل ذلك أخذ البيانات وتقسيمها بناءً على "الرموز المميزة" أو كلمات. بوجه عام، نزيل علامات الترقيم ونضبط جميع الكلمات على أحرف صغيرة في هذه الخطوة. لنفترض مثلاً أنّ لدينا السلسلة التالية: What time is it? بعد إنشاء الرمز المميّز، ستتألف هذه الجملة من أربعة رموز مميّزة: "what" , "time", "is", "it". لا نريد أن يتعامل النموذج مع الكلمة what ككلمتَين مختلفتَين بحرفتَين مختلفتَين بالأحرف اللاتينية الكبيرة. بالإضافة إلى ذلك، لا يساعدنا علامات الترقيم عادةً في تعلم الاستنتاج من الكلمات بشكل أفضل، لذلك تتم إزالتها أيضًا.

التسوية

غالبًا ما نريد "تسوية" البيانات. سيؤدي هذا إلى استبدال الكلمات ذات المعنى نفسه بالشيء نفسه. على سبيل المثال، إذا كانت الكلمات "قاتل" أو "قاتل" و"المبارزة" في النص، فقد يتم استبدال "تمت مواجهة" التسوية و"المبارزة" بكلمة "قتال".

الاشتقاق

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

إزالة كلمات الإيقاف

كلمات الإيقاف هي كلمات مثل "و" و"الـ" التي لا تضيف عادةً قيمة إلى المعنى الدلالي للجملة. عادةً ما نريد إزالتها كوسيلة لتقليل التشويش في مجموعات البيانات النصية الخاصة بنا.

8. إدارة المهمة

دعونا نلقي نظرة على الوظيفة التي نحن بصدد القيام بها. يمكنك العثور على الرمز على الرابط cloud-dataproc/codelabs/spark-nlp/topic_model.py. اقض على الأقل عدة دقائق في قراءة هذا التعليق والتعليقات المرتبطة به لفهم ما يحدث. سنسلّط الضوء أيضًا على بعض الأقسام أدناه:

# Python imports
import sys

# spark-nlp components. Each one is incorporated into our pipeline.
from sparknlp.annotator import Lemmatizer, Stemmer, Tokenizer, Normalizer
from sparknlp.base import DocumentAssembler, Finisher

# A Spark Session is how we interact with Spark SQL to create Dataframes
from pyspark.sql import SparkSession

# These allow us to create a schema for our data
from pyspark.sql.types import StructField, StructType, StringType, LongType

# Spark Pipelines allow us to sequentially add components such as transformers
from pyspark.ml import Pipeline

# These are components we will incorporate into our pipeline.
from pyspark.ml.feature import StopWordsRemover, CountVectorizer, IDF

# LDA is our model of choice for topic modeling
from pyspark.ml.clustering import LDA

# Some transformers require the usage of other Spark ML functions. We import them here
from pyspark.sql.functions import col, lit, concat

# This will help catch some PySpark errors
from pyspark.sql.utils import AnalysisException

# Assign bucket where the data lives
try:
    bucket = sys.argv[1]
except IndexError:
    print("Please provide a bucket name")
    sys.exit(1)

# Create a SparkSession under the name "reddit". Viewable via the Spark UI
spark = SparkSession.builder.appName("reddit topic model").getOrCreate()

# Create a three column schema consisting of two strings and a long integer
fields = [StructField("title", StringType(), True),
          StructField("body", StringType(), True),
          StructField("created_at", LongType(), True)]
schema = StructType(fields)

# We'll attempt to process every year / month combination below.
years = ['2016', '2017', '2018', '2019']
months = ['01', '02', '03', '04', '05', '06',
          '07', '08', '09', '10', '11', '12']

# This is the subreddit we're working with.
subreddit = "food"

# Create a base dataframe.
reddit_data = spark.createDataFrame([], schema)

# Keep a running list of all files that will be processed
files_read = []

for year in years:
    for month in months:

        # In the form of <project-id>.<dataset>.<table>
        gs_uri = f"gs://{bucket}/reddit_posts/{year}/{month}/{subreddit}.csv.gz"

        # If the table doesn't exist we will simply continue and not
        # log it into our "tables_read" list
        try:
            reddit_data = (
                spark.read.format('csv')
                .options(codec="org.apache.hadoop.io.compress.GzipCodec")
                .load(gs_uri, schema=schema)
                .union(reddit_data)
            )

            files_read.append(gs_uri)

        except AnalysisException:
            continue

if len(files_read) == 0:
    print('No files read')
    sys.exit(1)

# Replacing null values with their respective typed-equivalent is usually
# easier to work with. In this case, we'll replace nulls with empty strings.
# Since some of our data doesn't have a body, we can combine all of the text
# for the titles and bodies so that every row has useful data.

df_train = (
    reddit_data
    # Replace null values with an empty string
    .fillna("")
    .select(
         # Combine columns
        concat(
            # First column to concatenate. col() is used to specify that we're referencing a column
            col("title"),
            # Literal character that will be between the concatenated columns.
            lit(" "),
            # Second column to concatenate.
            col("body")
        # Change the name of the new column
        ).alias("text")
    )
)

# Now, we begin assembling our pipeline. Each component here is used to some transformation to the data.
# The Document Assembler takes the raw text data and convert it into a format that can
# be tokenized. It becomes one of spark-nlp native object types, the "Document".
document_assembler = DocumentAssembler().setInputCol("text").setOutputCol("document")

# The Tokenizer takes data that is of the "Document" type and tokenizes it.
# While slightly more involved than this, this is effectively taking a string and splitting
# it along ths spaces, so each word is its own string. The data then becomes the
# spark-nlp native type "Token".
tokenizer = Tokenizer().setInputCols(["document"]).setOutputCol("token")

# The Normalizer will group words together based on similar semantic meaning.
normalizer = Normalizer().setInputCols(["token"]).setOutputCol("normalizer")

# The Stemmer takes objects of class "Token" and converts the words into their
# root meaning. For instance, the words "cars", "cars'" and "car's" would all be replaced
# with the word "car".
stemmer = Stemmer().setInputCols(["normalizer"]).setOutputCol("stem")

# The Finisher signals to spark-nlp allows us to access the data outside of spark-nlp
# components. For instance, we can now feed the data into components from Spark MLlib.
finisher = Finisher().setInputCols(["stem"]).setOutputCols(["to_spark"]).setValueSplitSymbol(" ")

# Stopwords are common words that generally don't add much detail to the meaning
# of a body of text. In English, these are mostly "articles" such as the words "the"
# and "of".
stopword_remover = StopWordsRemover(inputCol="to_spark", outputCol="filtered")

# Here we implement TF-IDF as an input to our LDA model. CountVectorizer (TF) keeps track
# of the vocabulary that's being created so we can map our topics back to their
# corresponding words.
# TF (term frequency) creates a matrix that counts how many times each word in the
# vocabulary appears in each body of text. This then gives each word a weight based
# on its frequency.
tf = CountVectorizer(inputCol="filtered", outputCol="raw_features")

# Here we implement the IDF portion. IDF (Inverse document frequency) reduces
# the weights of commonly-appearing words.
idf = IDF(inputCol="raw_features", outputCol="features")

# LDA creates a statistical representation of how frequently words appear
# together in order to create "topics" or groups of commonly appearing words.
lda = LDA(k=10, maxIter=10)

# We add all of the transformers into a Pipeline object. Each transformer
# will execute in the ordered provided to the "stages" parameter
pipeline = Pipeline(
    stages = [
        document_assembler,
        tokenizer,
        normalizer,
        stemmer,
        finisher,
        stopword_remover,
        tf,
        idf,
        lda
    ]
)

# We fit the data to the model.
model = pipeline.fit(df_train)

# Now that we have completed a pipeline, we want to output the topics as human-readable.
# To do this, we need to grab the vocabulary generated from our pipeline, grab the topic
# model and do the appropriate mapping.  The output from each individual component lives
# in the model object. We can access them by referring to them by their position in
# the pipeline via model.stages[<ind>]

# Let's create a reference our vocabulary.
vocab = model.stages[-3].vocabulary

# Next, let's grab the topics generated by our LDA model via describeTopics(). Using collect(),
# we load the output into a Python array.
raw_topics = model.stages[-1].describeTopics().collect()

# Lastly, let's get the indices of the vocabulary terms from our topics
topic_inds = [ind.termIndices for ind in raw_topics]

# The indices we just grab directly map to the term at position <ind> from our vocabulary.
# Using the below code, we can generate the mappings from our topic indices to our vocabulary.
topics = []
for topic in topic_inds:
    _topic = []
    for ind in topic:
        _topic.append(vocab[ind])
    topics.append(_topic)

# Let's see our topics!
for i, topic in enumerate(topics, start=1):
    print(f"topic {i}: {topic}")

إدارة المهمة

لنمضي قدمًا وندير مهمتنا الآن. ابدأ وقم بتشغيل الأمر التالي:

gcloud dataproc jobs submit pyspark --cluster ${CLUSTER_NAME}\
    --region ${REGION}\
    --properties=spark.jars.packages=com.johnsnowlabs.nlp:spark-nlp_2.11:2.7.2\
    --driver-log-levels root=FATAL \
    topic_model.py \
    -- ${BUCKET_NAME}

يتيح لنا هذا الأمر الاستفادة من واجهة برمجة تطبيقات Dataproc Jobs API. من خلال تضمين الأمر pyspark، نشير إلى المجموعة بأنّ هذه مهمة PySpark. ونوفِّر اسم المجموعة والمَعلمات الاختيارية من تلك المتاحة هنا واسم الملف الذي يحتوي على المهمة. في هذه الحالة، نوفّر المَعلمة --properties التي تسمح لنا بتغيير الخصائص المختلفة لتطبيقات Spark أو Yarn أو Dataproc. نحن بصدد تغيير سمة Spark packages ما يتيح لنا إعلام Spark بأنّنا نريد تضمين spark-nlp ضمن مهمتنا. نوفّر أيضًا المعلَمات --driver-log-levels root=FATAL التي ستوقف معظم مخرجات السجلّ من PySpark باستثناء الأخطاء. بشكل عام، عادةً ما تكون سجلات Spark مزعجة.

أخيرًا، -- ${BUCKET} هي وسيطة سطر أوامر لنص Python البرمجي نفسه توفّر اسم الحزمة. يُرجى مراعاة المسافة بين -- و${BUCKET}.

بعد بضع دقائق من تشغيل المهمة، من المفترض أن تظهر لنا نتيجة تحتوي على النماذج التالية:

رائع. هل يمكنك استنتاج الاتجاهات من خلال النظر إلى الناتج من النموذج؟ ماذا عن شركتنا؟

من الناتج أعلاه، قد يستنتج المرء اتجاهًا من الموضوع 8 فيما يتعلق بطعام الإفطار والحلويات من الموضوع 9.

9. تنظيف

لتجنُّب تحمُّل أي رسوم غير ضرورية من حسابك على Google Cloud Platform بعد إكمال عملية البدء السريع هذه:

1. حذف حزمة Cloud Storage للبيئة التي أنشأتها
2. احذف بيئة Dataproc.

إذا أنشأت مشروعًا لهذا الدرس التطبيقي حول الترميز فقط، يمكنك أيضًا حذفه اختياريًا:

1. في "وحدة تحكّم Google Cloud Platform"، انتقِل إلى صفحة المشاريع.
2. في قائمة المشاريع، اختَر المشروع الذي تريد حذفه وانقر على حذف.
3. في المربع، اكتب رقم تعريف المشروع، ثم انقر على إيقاف التشغيل لحذف المشروع.

تحذير: يؤدي حذف المشروع إلى التأثيرات التالية:

• تم حذف كل محتوى المشروع. إذا استخدمت مشروعًا حاليًا لهذا البرنامج التعليمي، فعند حذفه، يتم أيضًا حذف أي عمل آخر قمت به في هذا المشروع.
• يتم فقدان أرقام تعريف المشاريع المخصّصة. عندما أنشأت هذا المشروع، ربما تكون قد أنشأت رقم تعريف مشروع مخصّصًا تريد استخدامه في المستقبل. للاحتفاظ بعناوين URL التي تستخدم رقم تعريف المشروع، مثل عنوان URL يتضمّن appspot.com، عليك حذف الموارد المحدَّدة داخل المشروع بدلاً من حذف المشروع بالكامل.

الترخيص

هذا العمل مرخّص بموجب ترخيص Creative Commons Attribution 3.0 العام وترخيص Apache 2.0.