<jupyter_start><jupyter_text>Tutorial: Sending PySpark DataFrame to ArizeIn the current version of Arize Python SDK, only Pandas DataFrames are supported. To log Spark DataFrames, which have `rdds` as their underlying structure, we will use `mapPartitions` to log them to arize.  Install Dependencies in Colab<jupyter_code>!pip install -q pyspark
!pip install -q arize<jupyter_output><empty_output><jupyter_text>Parallelizing PySpark DataFrameWe first create a dummy PySpark DataFrame to send.<jupyter_code>import pyspark
from pyspark.sql import Row, SparkSession
import pandas as pd

spark = SparkSession.builder.getOrCreate()

# Read some dummy data for logging to Arize later
data = pd.read_csv(
    "https://storage.googleapis.com/arize-assets/fixtures/compare-model-a.csv"
)
df_pandas = data[["loan_amount", "interest_rate", "grade", "prediction", "score"]]

df_pandas = pd.concat([df_pandas] * 5)

print("This is a pandas DataFrame:")
display(df_pandas)

# Create PySpark dataframe unparallelized
df_spark = spark.createDataFrame(df_pandas)

print("\nThis is the corresponding spark DataFrame")
df_spark.printSchema()<jupyter_output><empty_output><jupyter_text>Using `mapPartitions` to log each partition to Arize`map_func` will be applied to each partition on of spark_df, allowing local copies of `pd.DataFrame` to be made and logged to Arize.`success` will be returned if all entries in a particular partition has been logged properly. Otherwise, it will return the error code and error message for that partition instead.`map_send_arize` should send `spark_df` to Arize with at least one of: `shap, prediction_labels, actual_labels` How To Log to Arize:`your_spark_df.rdd.mapPartitions(map_func).collect()`You will also need to update the `API_KEY` and `ORGANIZATION_KEY` Setting up Arize Client:First, copy the Arize `API_KEY` and `ORG_KEY` from your admin page linked below![![Button_Open.png](https://storage.googleapis.com/arize-assets/fixtures/Button_Open.png)](https://app.arize.com/admin)<jupyter_code>from arize.pandas.logger import Client, Schema
from arize.utils.types import ModelTypes, Environments

ORGANIZATION_KEY = "ORGANIZATION_KEY"
API_KEY = "API_KEY"
arize_client = Client(organization_key=ORGANIZATION_KEY, api_key=API_KEY)

if ORGANIZATION_KEY == "ORGANIZATION_KEY" or API_KEY == "API_KEY":
    raise ValueError("❌ NEED TO CHANGE ORGANIZATION AND/OR API_KEY")<jupyter_output><empty_output><jupyter_text>Define the function used by `pyspark.RDD.mapPartitions`<jupyter_code>import itertools
import uuid


def map_send_arize(pyspark_df_partition: itertools.chain):
    """
    Mapping function to be used to log to Arize
    """
    # Step 1: iterating through each Row to create pd.DataFrame
    pandas_df = None
    for row in pyspark_df_partition:
        row_dict = row.asDict()
        row_dict["prediction_id"] = str(uuid.uuid4())
        if pandas_df is None:
            pandas_df = pd.DataFrame(columns=row_dict.keys())
        pandas_df.loc[len(pandas_df)] = row_dict

    # Step 2: Define a Schema() object for Arize to pick up data from the correct columns for logging
    pandas_df_schema = Schema(
        prediction_id_column_name="prediction_id",  # REQUIRED
        prediction_label_column_name="prediction",
        prediction_score_column_name="score",
        feature_column_names=["loan_amount", "interest_rate", "grade"],
    )

    # Step 3: Log to arize
    response = arize_client.log(
        dataframe=pandas_df,
        schema=pandas_df_schema,
        model_id="pyspark-loan-model",
        model_version="1.0",
        model_type=ModelTypes.SCORE_CATEGORICAL,
        environment=Environments.PRODUCTION,
    )

    # Step 4: Check for errors when logging
    return iter([f"Status {response.status_code}: {response.text}"])<jupyter_output><empty_output><jupyter_text>Logging ExampleFor each `log` response we get from Arize, `collect()` will combine the iterables as the map function completes.<jupyter_code>%%time
log_output = df_spark.rdd.mapPartitions(map_send_arize).collect()
print(f"number of records sent: {len(df_pandas)}")
print(f"number of arize responses: {len(log_output)}")
if all(response.startswith("Status 200:") for response in log_output):
    print(f"✅ all log requests are successful")
else:
    for response in log_output:
        if not response.startswith("Status 200:"):
            print(f"❌ {response}")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09529198732458126, 'fraction_numerical': 0.003168854685377999, 'mean_word_length': 5.070054945054945, 'pattern_counts': {'":': 1, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '650083', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 1471, 'n_tokens_neox': 1426, 'n_words': 392, 'path': 'arize/examples/tutorials/Arize_Tutorials/Log_Examples/Arize_Tutorial_PySpark_V1.ipynb', 'repo_name': 'Arize-ai/client_python', 'size': '4418'}