Skip to content

Pandas UDF Failure #435

Description

@khageshbatra

Describe the bug

When trying to create a PANDAS UDF to run the latest GPT3.5Turbo API via Azure openAI completion function it gives an error regarding missing key.

To Reproduce

Code snippets

#!pip install openai

from pyspark.sql import SparkSession
import openai
from pyspark.sql.functions import pandas_udf  
from pyspark.sql.types import StringType
import pandas as pd
import os

# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()

service_name = "oai-eus-ww-int-hqfin"
deployment_name = "GPT35Model"

os.environ["OPENAI_API_KEY"] = ""

# Setup for OpenAI API for GPT3.5Turbo
openai.api_type = "azure"
openai.api_base = ""
openai.api_version = "2023-03-15-preview"
openai.api_key = os.environ.get("OPENAI_API_KEY")

# Create a Spark DataFrame with a prompt column
data = [  
    ("Python code for Hello World"),  
    ("What is GZCL?"),
    ("Who is Sandman?")  
]  
df = spark.createDataFrame(pd.DataFrame(data, columns=["prompt"]))

# Define the pandas UDF function  
@pandas_udf(returnType=StringType())
def generate_text(input: pd.Series) -> pd.Series:  
    results = []
    prompt = ""
    for n in input:  
        message = prompt + n  
          
        responseGPT = openai.ChatCompletion.create(  
            engine="GPT35Model",  
            messages=[{  
                "role": "user",  
                "content": message  
            }],  
            temperature=0.1,  
            max_tokens=800,  
            top_p=0.95,  
            frequency_penalty=0,  
            presence_penalty=0,  
            stop=None  
        )  
          
        results.append(responseGPT["choices"][0]["message"]["content"])  
      
    return pd.Series(results)

df = df.withColumn("results", generate_text("prompt")).cache()
display(df)

OS

linux

Python version

Python v3.10

Library version

openai-python v0.27.6

Activity

  1. sreedhar-guda commented on Aug 22, 2023

    @sreedhar-guda

    Similar issue noticed with openai.Embedding.create as well. No such issues when running it like a regular function.

    Possible workaround:
    Set OpenAI related environment variables before openai import statement as shown below:

    import os
    os.environ["OPENAI_API_KEY"]=<api-key-value> 
    os.environ["OPENAI_API_TYPE"]="azure"
    os.environ["OPENAI_API_BASE"]="https://<your-resource-name>.openai.azure.com/"
    os.environ["OPENAI_API_VERSION"]="2023-05-15"
    
    import openai
    
    rest of the code....
    
  2. vandustry commented on Oct 17, 2023

    @vandustry

    This may be loosely related (but if you don't necessarily need to use pandas, it may solve the problem):

    I had similar issues with pyspark.sql.functions.udf (running it as a regular function works, inside a udf not). Unfortunately, the work-around suggested by @sreedhar-guda did not work for me in this case. However, I think I figured out the issue/origin:

    udf uses workers with their own environment variables. That means, you have to broadcast "global" variables to their environment. You can do this using spark.sparkContext.broadcast. Here is what it would look like:

    # Define broadcast-variables.
    broadcast_api_key = spark.sparkContext.braodcast(os.environ.get("OPENAI_API_KEY"))
    broadcast_api_type = spark.sparkContext.braodcast("azure")
    broadcast_api_base = spark.sparkContext.braodcast("")
    broadcast_api_version = spark.sparkContext.braodcast("2023-03-15-preview")
    
    # Define the UDF function  
    def generate_text(input):  
        openai.api_key = broadcast_api_key.value
        openai.api_type = broadcast_api_type.value
        openai.api_base = broadcast_api_base.value
        openai.api_version = broadcast_api_version.value
    
        prompt = ""
        message = prompt + n  
              
        responseGPT = openai.ChatCompletion.create(  
            engine="GPT35Model",  
            messages=[{  
                "role": "user",  
                "content": message  
            }],  
            temperature=0.1,  
            max_tokens=800,  
            top_p=0.95,  
            frequency_penalty=0,  
            presence_penalty=0,  
            stop=None  
        )  
              
        return responseGPT["choices"][0]["message"]["content"]
          
    df = df.withColumn("results", udf(generate_text, StringType())(col(column_name)))
    
  3. rattrayalex commented on Nov 10, 2023

    @rattrayalex
    Contributor

    We recently rewrote the library entirely; if the problem persists in v1, please open a new issue (ideally with a replit that can reproduce the problem).

  4. added a commit that references this issue on Apr 27, 2025
    616d8e7
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't working

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions