PySpark untuk Natural Language Processing di Dataproc

PySpark untuk Natural Language Processing di Dataproc

Tentang codelab ini

Terakhir diperbarui Jun 25, 2021

Ditulis oleh bmiro

1. Ringkasan

Natural Language Processing (NLP) adalah studi untuk mendapatkan insight dan melakukan analisis pada data tekstual. Seiring bertambahnya jumlah tulisan yang dihasilkan di internet, kini makin banyak organisasi yang berupaya memanfaatkan teks untuk mendapatkan informasi yang relevan dengan bisnis mereka.

NLP dapat digunakan untuk apa saja, mulai dari menerjemahkan bahasa, menganalisis sentimen, hingga membuat kalimat dari awal, dan banyak lagi. Ini adalah area penelitian aktif yang mengubah cara kita bekerja dengan teks.

Kita akan mengeksplorasi cara menggunakan NLP pada data tekstual dalam jumlah besar dalam skala besar. Hal ini tentu menjadi tugas yang sulit! Untungnya, kita akan memanfaatkan library seperti Spark MLlib dan spark-nlp untuk mempermudah hal ini.

2. Kasus Penggunaan Kami

{i>Chief Data Scientist<i} dari organisasi (fiksi) kami, "FoodCorp" tertarik untuk mempelajari lebih lanjut tentang tren dalam industri makanan. Kita memiliki akses ke korpus data teks dalam bentuk postingan dari subreddit r/food Reddit yang akan digunakan untuk mengeksplorasi apa yang sedang dibicarakan orang.

Salah satu pendekatan untuk melakukan ini adalah melalui metode NLP yang dikenal sebagai "pemodelan topik". Pemodelan topik adalah metode statistik yang dapat mengidentifikasi tren dalam makna semantik dari sekelompok dokumen. Dengan kata lain, kita bisa membangun model topik di korpus "postingan" Reddit yang akan menghasilkan daftar "topik" atau kelompok kata yang menggambarkan suatu tren.

Untuk membangun model, kita akan menggunakan algoritma yang disebut Latent Dirichlet Allocation (LDA), yang sering digunakan untuk mengelompokkan teks. Pengantar LDA yang sangat baik dapat ditemukan di sini.

3. Membuat Project

Jika belum memiliki Akun Google (Gmail atau Google Apps), Anda harus membuatnya. Login ke konsol Google Cloud Platform ( console.cloud.google.com) dan buat project baru:

Selanjutnya, Anda harus mengaktifkan penagihan di Cloud Console untuk menggunakan resource Google Cloud.

Menjalankan operasi dalam codelab ini seharusnya tidak menghabiskan biaya lebih dari beberapa dolar, tetapi bisa lebih mahal jika Anda memutuskan untuk menggunakan lebih banyak resource atau jika Anda membiarkannya berjalan. Codelab PySpark-BigQuery dan Spark-NLP menjelaskan "Pembersihan" di akhir.

Pengguna baru Google Cloud Platform memenuhi syarat untuk mendapatkan uji coba gratis senilai$300.

4. Menyiapkan Lingkungan Kita

Pertama, kita harus mengaktifkan Dataproc dan Compute Engine API.

Klik ikon menu di kiri atas layar.

Pilih Pengelola API dari menu drop-down.

Klik Enable APIs and Services.

Telusuri "Compute Engine" di kotak penelusuran. Klik "Google Compute Engine API" di daftar hasil yang muncul.

Di halaman Google Compute Engine, klik Enable

Setelah diaktifkan, klik tanda panah yang menunjuk ke kiri untuk kembali.

Sekarang telusuri "Google Dataproc API" dan mengaktifkannya juga.

Selanjutnya, buka Cloud Shell dengan mengklik tombol di pojok kanan atas Cloud Console:

Kita akan menetapkan beberapa variabel lingkungan yang dapat direferensikan saat melanjutkan codelab. Pertama, pilih nama untuk cluster Dataproc yang akan dibuat, misalnya "my-cluster", lalu tetapkan di lingkungan Anda. Jangan ragu untuk menggunakan nama apa pun yang Anda inginkan.

CLUSTER_NAME=my-cluster

Selanjutnya, pilih zona dari salah satu zona yang tersedia di sini. Contohnya adalah us-east1-b.

REGION=us-east1

Terakhir, kita perlu menetapkan bucket sumber yang akan dibaca oleh tugas kita. Kami memiliki data sampel yang tersedia di bucket bm_reddit, tetapi Anda dapat menggunakan data yang dihasilkan dari PySpark untuk Data BigQuery untuk Prapemrosesan jika Anda menyelesaikannya sebelumnya.

BUCKET_NAME=bm_reddit

Setelah variabel lingkungan telah dikonfigurasi, jalankan perintah berikut untuk membuat cluster Dataproc:

 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

Mari kita lakukan setiap perintah berikut:

gcloud beta dataproc clusters create ${CLUSTER_NAME}: akan memulai pembuatan cluster Dataproc dengan nama yang Anda berikan sebelumnya. Kami menyertakan beta di sini untuk mengaktifkan fitur beta Dataproc seperti Gateway Komponen, yang akan kita bahas di bawah ini.

--zone=${ZONE}: Tindakan ini akan menetapkan lokasi cluster.

--worker-machine-type n1-standard-8: Ini adalah jenis mesin yang digunakan untuk pekerja kita.

--num-workers 4: Kami akan memiliki empat worker di cluster.

--image-version 1.4-debian9: Kode ini menunjukkan versi gambar Dataproc yang akan kita gunakan.

--initialization-actions ...: Tindakan Inisialisasi adalah skrip kustom yang dijalankan saat membuat cluster dan worker. Aset ini dapat dibuat oleh pengguna dan disimpan di bucket GCS atau dirujuk dari bucket publik dataproc-initialization-actions. Tindakan inisialisasi yang disertakan di sini akan memungkinkan penginstalan paket Python menggunakan Pip, seperti yang diberikan dengan flag --metadata.

--metadata 'PIP_PACKAGES=google-cloud-storage spark-nlp': Ini adalah daftar paket yang dipisahkan spasi untuk diinstal ke Dataproc. Dalam hal ini, kita akan menginstal library klien Python google-cloud-storage dan spark-nlp.

--optional-components=ANACONDA: Komponen Opsional adalah paket umum yang digunakan dengan Dataproc, yang diinstal otomatis di cluster Dataproc selama pembuatan. Keuntungan menggunakan Komponen Opsional dibandingkan Tindakan Inisialisasi mencakup waktu startup yang lebih cepat dan pengujian untuk versi Dataproc tertentu. Secara keseluruhan, aset tersebut lebih dapat diandalkan.

--enable-component-gateway: Flag ini memungkinkan kita memanfaatkan Gateway Komponen Dataproc untuk melihat UI umum seperti Zeppelin, Jupyter, atau Histori Spark. Catatan: beberapa di antaranya memerlukan Komponen Opsional terkait.

Untuk pengantar Dataproc yang lebih mendalam, lihat codelab ini.

Selanjutnya, jalankan perintah berikut di Cloud Shell untuk meng-clone repo dengan kode sampel dan menjalankan cd ke direktori yang benar:

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

5. Spark MLlib

Spark MLlib adalah library machine learning skalabel yang ditulis di Apache Spark. Dengan memanfaatkan efisiensi Spark menggunakan serangkaian algoritma machine learning yang telah disesuaikan, MLlib dapat menganalisis data dalam jumlah besar. Memiliki API di Java, Scala, Python, dan R. Dalam codelab ini, kita akan secara khusus berfokus pada Python.

MLlib berisi serangkaian besar transformer dan estimator. Transformer adalah alat yang dapat mengubah atau mengubah data Anda, biasanya dengan fungsi transform(), sedangkan estimator adalah algoritma bawaan yang dapat digunakan untuk melatih data Anda, biasanya dengan fungsi fit().

Contoh transformer meliputi:

• tokenization (membuat vektor angka dari string kata)
• encoding one-hot (membuat vektor angka yang jarang yang mewakili kata-kata yang ada dalam string)
• remover stopwords (menghapus kata yang tidak menambahkan nilai semantik ke string)

Contoh estimator mencakup:

• klasifikasi (apakah ini apel atau jeruk?)
• regresi (berapa seharusnya harga apel ini?)
• pengelompokan (seberapa mirip semua apel satu sama lain?)
• pohon keputusan (jika warna == oranye, maka itu adalah oranye. Jika tidak, itu adalah apel)
• pengurangan dimensi (dapatkah kita menghapus fitur dari {i>dataset<i} kita dan masih membedakan antara apel dan jeruk?).

MLlib juga berisi alat untuk metode umum lainnya dalam machine learning seperti penyesuaian dan pemilihan hyperparameter serta validasi silang.

Selain itu, MLlib berisi Pipelines API, yang memungkinkan Anda membangun pipeline transformasi data menggunakan berbagai transformer yang dapat dieksekusi kembali.

6. Spark-NLP

Spark-nlp adalah library yang dibuat oleh John Snow Labs untuk melakukan tugas natural language processing yang efisien menggunakan Spark. Library ini berisi alat bawaan yang disebut anotator untuk tugas umum seperti:

• tokenization (membuat vektor angka dari string kata)
• membuat embedding kata (menentukan hubungan antara kata-kata melalui vektor)
• tag part-of-speech (kata apa yang merupakan kata benda? Apa yang dimaksud dengan kata kerja?)

Meskipun berada di luar cakupan codelab ini, spark-nlp juga terintegrasi dengan bagus dengan TensorFlow.

Mungkin yang paling signifikan, Spark-NLP memperluas kemampuan Spark MLlib dengan menyediakan komponen yang mudah dimasukkan ke dalam Pipeline MLlib.

7. Praktik Terbaik untuk Natural Language Processing

Sebelum kita bisa mengekstrak informasi yang berguna dari data, kita perlu mengurus beberapa hal kecil. Langkah-langkah pra-pemrosesan yang akan kita lakukan adalah sebagai berikut:

Tokenisasi

Hal pertama yang biasanya ingin kita lakukan adalah "tokenize" yang mengupload data. Ini melibatkan pengambilan data dan memisahkannya berdasarkan "token" atau kata-kata. Biasanya, kami menghapus tanda baca dan menetapkan semua kata ke huruf kecil dalam langkah ini. Misalnya, kita memiliki string berikut: What time is it? Setelah tokenisasi, kalimat ini akan terdiri dari empat token: "what" , "time", "is", "it". Kita tidak ingin model memperlakukan kata what sebagai dua kata yang berbeda dengan dua huruf besar yang berbeda. Selain itu, tanda baca biasanya tidak membantu kita mempelajari inferensi dengan lebih baik dari kata-kata, jadi kita juga menghapusnya.

Normalisasi

Kita sering melakukan "normalisasi" yang mengupload data. Hal ini akan mengganti kata-kata yang memiliki makna serupa dengan hal yang sama. Misalnya, jika kata "berjuang", "berjuang" dan "berduet" diidentifikasi dalam teks, maka normalisasi dapat menggantikan "battled" dan "berduet" dengan kata "bertarung".

Bentuk dasar

Stemmer akan mengganti kata dengan arti akarnya. Misalnya, kata "mobil", "mobil" dan "mobil" semuanya akan diganti dengan kata "{i>car<i}", karena semua kata tersebut menyiratkan hal yang sama.

Menghapus Stopword

Stopword adalah kata-kata seperti "dan" dan "the" yang biasanya tidak memberikan nilai tambah terhadap makna semantik sebuah kalimat. Kita biasanya ingin menghapusnya sebagai cara untuk mengurangi derau pada set data teks.

8. Menjalankan Pekerjaan

Mari kita lihat tugas yang akan kita jalankan. Kode ini dapat ditemukan di cloud-dataproc/codelabs/spark-nlp/topic_model.py. Luangkan setidaknya beberapa menit untuk membacanya dan komentar terkait untuk memahami apa yang terjadi. Kami juga akan menyoroti beberapa bagian di bawah ini:

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

Menjalankan Tugas

Sekarang mari kita lanjutkan dan jalankan tugas kita. Lanjutkan dan jalankan perintah berikut:

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}

Perintah ini memungkinkan kita memanfaatkan Dataproc Jobs API. Dengan menyertakan perintah pyspark, kita akan menunjukkan ke cluster bahwa ini adalah tugas PySpark. Kami menyediakan nama cluster, parameter opsional dari yang tersedia di sini, dan nama file yang berisi tugas. Dalam kasus ini, kita memberikan parameter --properties yang memungkinkan kita mengubah berbagai properti untuk Spark, Yarn, atau Dataproc. Kita mengubah properti Spark packages yang memungkinkan kita memberi tahu Spark bahwa kita ingin menyertakan spark-nlp sebagai paket dengan tugas kita. Kami juga menyediakan parameter --driver-log-levels root=FATAL yang akan menyembunyikan sebagian besar output log dari PySpark kecuali untuk Error. Secara umum, log Spark cenderung berisik.

Terakhir, -- ${BUCKET} adalah argumen command line untuk skrip Python itu sendiri yang memberikan nama bucket. Perhatikan spasi antara -- dan ${BUCKET}.

Setelah beberapa menit menjalankan tugas, kita akan melihat output yang berisi model kita:

Mengagumkan!! Dapatkah Anda menyimpulkan tren dengan melihat output dari model Anda? Bagaimana dengan punya kita?

Dari output di atas, kita dapat menyimpulkan tren dari topik 8 yang berkaitan dengan makanan sarapan, dan makanan penutup dari topik 9.

9. Pembersihan

Agar tidak menimbulkan tagihan yang tidak perlu pada akun GCP Anda setelah menyelesaikan panduan memulai ini:

1. Hapus bucket Cloud Storage untuk lingkungan dan yang Anda buat
2. Menghapus lingkungan Dataproc.

Jika membuat project hanya untuk codelab ini, Anda juga dapat menghapus project tersebut secara opsional:

1. Di GCP Console, buka halaman Project.
2. Dalam daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
3. Pada kotak, ketik project ID, lalu klik Shut down untuk menghapus project.

Perhatian: Menghapus project memiliki efek berikut:

• Semua hal dalam project akan dihapus. Jika Anda menggunakan project yang ada untuk tutorial ini, saat menghapusnya, Anda juga akan menghapus semua pekerjaan lain yang telah Anda lakukan di project tersebut.
• Project ID kustom akan hilang. Saat membuat project ini, Anda mungkin telah membuat project ID kustom yang masih ingin Anda gunakan pada masa mendatang. Untuk mempertahankan URL yang menggunakan project ID, seperti URL appspot.com, hapus resource yang dipilih di dalam project, bukan menghapus keseluruhan project.

Lisensi

Karya ini dilisensikan berdasarkan Lisensi Generik Creative Commons Attribution 3.0 dan Apache 2.0.