Analyse text sentiment with Azure and PySpark

This 2020 tutorial used Azure Cognitive Services Text Analytics v3.0 from a Databricks PySpark notebook. It sent tweet text to the sentiment endpoint, parsed the JSON response, and joined the scores back to the source data.

Azure product names, endpoints, SDKs, supported languages, and authentication guidance have changed. Use the flow as a historical reference and Microsoft’s current sentiment and opinion-mining documentation for a new build.

The original data flow

The notebook followed eight steps:

1. load a dataset with an identifier and text field
2. select the text field to analyse
3. configure the service endpoint and credential
4. shape records into the API document format
5. serialize the batch to JSON
6. send the request
7. flatten the response
8. join sentiment values back to the source rows

The example used a COVID-19 tweet dataset.

Load and shape the source data

The original notebook loaded a CSV and removed records without text:

from pyspark.sql.functions import col, monotonically_increasing_id

source = (
    spark.read.format("csv")
    .option("inferSchema", "false")
    .option("header", "true")
    .load("/FileStore/tables/TwitterCovidTweets.csv")
)

text_column = "Text"
prepared = (
    source.filter(col(text_column).isNotNull())
    .withColumn("ID", monotonically_increasing_id())
    .selectExpr("ID", "Language", text_column)
)

A production design should use a stable source identifier rather than create one only for the current Spark execution.

Build the request without exposing credentials

The 2020 post placed the endpoint and subscription key directly in notebook variables. Do not copy that pattern.

Use a managed identity or an approved secret store where the current service and runtime support it. Limit the identity to the required resource and keep secrets out of notebook output, source control, and telemetry.

The archived request shape was:

import json
import requests

records = [json.loads(row) for row in prepared.toJSON().collect()]
document = {"documents": records}

headers = {"Ocp-Apim-Subscription-Key": subscription_key}
response = requests.post(endpoint, headers=headers, json=document)
response.raise_for_status()
payload = response.json()

Collecting all rows to the driver does not scale to an unrestricted dataset. Current implementations should batch within documented service limits and handle retry, throttling, partial failure, and data-location requirements.

Inspect and flatten the response

The API returned a result for each document identifier.

The original notebook converted the returned documents into a Spark DataFrame and selected sentiment and confidence values:

from pyspark.sql.functions import col

results = spark.read.json(
    spark.sparkContext.parallelize(payload["documents"])
)

scores = results.select(
    col("id").alias("ID"),
    col("sentiment").alias("Sentiment"),
    col("confidenceScores.positive").alias("Positive"),
    col("confidenceScores.neutral").alias("Neutral"),
    col("confidenceScores.negative").alias("Negative"),
)

final = prepared.join(scores, prepared.ID == scores.ID, "left").drop(scores.ID)

Validate before interpreting the result

A sentiment score is a model output, not a fact about the customer or writer. Test the current model against representative language, domain terms, sarcasm, short text, and the languages present in the dataset.

Record the service version, input policy, evaluation set, and intended use. Do not use a general sentiment label as an automated decision about a person without a separate risk and governance review.