# Create EMR cluster with Sparkaws emr create-cluster \  --name "GEO-ChatGPT-Simulation" \  --release-label emr-6.15.0 \  --applications Name=Spark Name=Hadoop \  --instance-type m5.xlarge \  --instance-count 11 \  --use-default-roles \  --ec2-attributes KeyName=your-key-pair

# Install on EMR cluster!pip install openai pandas boto3 pyspark# Import necessary librariesfrom pyspark.sql import SparkSessionfrom pyspark.sql.functions import *from pyspark.sql.types import *import openaiimport jsonimport timefrom concurrent.futures import ThreadPoolExecutor

# Configure Spark for optimal API processingspark = SparkSession.builder \    .appName("ChatGPT-GEO-Simulation") \    .config("spark.sql.adaptive.enabled", "true") \    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \    .getOrCreate()# Set optimal partition count for API callsspark.conf.set("spark.sql.shuffle.partitions", "200")

# Define schema for Relixir query exportquery_schema = StructType([    StructField("query_id", StringType(), True),    StructField("query_text", StringType(), True),    StructField("query_category", StringType(), True),    StructField("buyer_stage", StringType(), True),    StructField("priority_score", IntegerType(), True),    StructField("competitor_keywords", ArrayType(StringType()), True)])# Load Relixir query exportqueries_df = spark.read \    .option("header", "true") \    .schema(query_schema) \    .csv("s3://your-bucket/relixir-queries.csv")print(f"Loaded {queries_df.count()} queries for simulation")

# Categorize queries by buyer journey stagequery_categories = queries_df.groupBy("buyer_stage").count().show()# Sample output:# +-------------+-----+# |  buyer_stage|count|# +-------------+-----+# |    awareness| 3247|# |consideration| 4156|# |    decision | 2597

# Configure OpenAI client with rate limitingopenai.api_key = "your-openai-api-key"def execute_chatgpt_query(query_text, max_retries=3):    """Execute single ChatGPT query with retry logic"""    for attempt in range(max_retries):        try:            response = openai.ChatCompletion.create(                model="gpt-4",                messages=[                    {"role": "user", "content": query_text}                ],                max_tokens=500,                temperature=0.1  # Consistent responses for analysis            )                        return {                "response_text": response.choices[0].message.content,                "tokens_used": response.usage.total_tokens,                "model": response.model,                "timestamp": time.time()            }                    except Exception as e:            if attempt == max_retries - 1:                return {"error": str(e), "timestamp": time.time()}            time.sleep(2 ** attempt)  # Exponential backoff

# Create UDF for parallel API callsfrom pyspark.sql.functions import udffrom pyspark.sql.types import MapType, StringType# Define UDF for ChatGPT API callschatgpt_udf = udf(execute_chatgpt_query, MapType(StringType(), StringType()))# Execute queries in parallel across Spark clusterresults_df = queries_df.withColumn(    "chatgpt_response",     chatgpt_udf(col("query_text")))# Cache results for iterative analysisresults_df.cache()print(f"Processed {results_df.count()} ChatGPT queries")

# Process queries in optimized batchesdef process_query_batch(batch_size=100):    """Process queries in batches to optimize API usage"""        # Partition queries by priority score    high_priority = queries_df.filter(col("priority_score") >= 8)    medium_priority = queries_df.filter(        (col("priority_score") >= 5) & (col("priority_score") < 8)    )    low_priority = queries_df.filter(col("priority_score") < 5)        # Process high priority queries first    for priority_df, priority_name in [        (high_priority, "high"),        (medium_priority, "medium"),         (low_priority, "low")    ]:        print(f"Processing {priority_name} priority queries...")                batch_results = priority_df.withColumn(            "chatgpt_response",            chatgpt_udf(col("query_text"))        )                # Save intermediate results        batch_results.write \            .mode("overwrite") \            .parquet(f"s3://your-bucket/results/{priority_name}_priority/")

# Extract brand mentions from ChatGPT responsesdef extract_brand_mentions(response_text, brand_list):    """Extract brand mentions and sentiment from response"""    mentions = []    response_lower = response_text.lower()        for brand in brand_list:        if brand.lower() in response_lower:            # Find position and context            position = response_lower.find(brand.lower())            context = response_text[max(0, position-50):position+len(brand)+50]                        mentions.append({                "brand": brand,                "position": position,                "context": context,                "mentioned": True            })        return mentions# Apply brand extraction across all responsesbrand_list = ["YourBrand", "Competitor1", "Competitor2", "Competitor3"]brand_mentions_udf = udf(    lambda text: extract_brand_mentions(text, brand_list),    ArrayType(MapType(StringType(), StringType())))analysis_df = results_df.withColumn(    "brand_mentions",    brand_mentions_udf(col("chatgpt_response.response_text")))

# Analyze competitive positioning across query categoriescompetitive_analysis = analysis_df \    .select(        "query_category",        "buyer_stage",         explode("brand_mentions").alias("mention")    ) \    .select(        "query_category",        "buyer_stage",        col("mention.brand").alias("brand"),        col("mention.position").alias("mention_position")    ) \    .groupBy("query_category", "buyer_stage", "brand") \    .agg(        count("*").alias("mention_count"),        avg("mention_position").alias("avg_position")    ) \    .orderBy("query_category", "buyer_stage", desc("mention_count"))competitive_analysis.show(50)

# Score response quality and relevancedef calculate_response_score(response_text, query_text):    """Calculate response quality score based on multiple factors"""    if not response_text or "error" in response_text:        return 0        # Length score (optimal range: 100-400 words)    word_count = len(response_text.split())    length_score = min(1.0, word_count / 200) if word_count < 200 else max(0.5, 400 / word_count)        # Relevance score (keyword overlap)    query_words = set(query_text.lower().split())    response_words = set(response_text.lower().split())    relevance_score = len(query_words.intersection(response_words)) / len(query_words)        # Structure score (presence of lists, examples)    structure_score = 0.5    if any(marker in response_text for marker in ["1.", "2.", "-", "•"]):        structure_score += 0.3    if any(word in response_text.lower() for word in ["example", "for instance", "such as"]):        structure_score += 0.2        return (length_score * 0.4 + relevance_score * 0.4 + structure_score * 0.2) * 100response_score_udf = udf(calculate_response_score, FloatType())scored_df = analysis_df.withColumn(    "response_quality_score",    response_score_udf(        col("chatgpt_response.response_text"),        col("query_text")    ))

# Prepare results for Relixir dashboard integrationdashboard_export = scored_df.select(    "query_id",    "query_text",    "query_category",    "buyer_stage",    "priority_score",    col("chatgpt_response.response_text").alias("ai_response"),    col("chatgpt_response.tokens_used").alias("tokens_consumed"),    "brand_mentions",    "response_quality_score",    current_timestamp().alias("processed_at"))# Export to S3 for Relixir ingestiondashboard_export.write \    .mode("overwrite") \    .option("header", "true") \    .csv("s3://your-bucket/relixir-dashboard-import/")print(f"Exported {dashboard_export.count()} processed queries to Relixir dashboard")

# Calculate key GEO performance metricsperformance_metrics = scored_df.agg(    # Brand visibility metrics    (sum(when(array_contains(col("brand_mentions.brand"), "YourBrand"), 1).otherwise(0)) / count("*") * 100).alias("brand_mention_rate"),        # Response quality metrics      avg("response_quality_score").alias("avg_response_quality"),        # Competitive metrics    (sum(when(array_contains(col("brand_mentions.brand"), "YourBrand"), 1).otherwise(0)) /      sum(when(size(col("brand_mentions")) > 0, 1).otherwise(0)) * 100).alias("competitive_share")).collect()[0]print(f"Brand Mention Rate: {performance_metrics['brand_mention_rate']:.2f}%")print(f"Average Response Quality: {performance_metrics['avg_response_quality']:.2f}")print(f"Competitive Share: {performance_metrics['competitive_share']:.2f}%")

# Identify content gaps where competitors dominatecontent_gaps = scored_df \    .filter(~array_contains(col("brand_mentions.brand"), "YourBrand")) \    .filter(size(col("brand_mentions")) > 0) \    .groupBy("query_category", "buyer_stage") \    .agg(        count("*").alias("gap_count"),        collect_list("query_text").alias("gap_queries"),        avg("response_quality_score").alias("competitor_avg_quality")    ) \    .orderBy(desc("gap_count"))content_gaps.show(truncate=False)

# Generate content briefs for identified gapsdef generate_content_brief(query_category, gap_queries, competitor_responses):    """Generate actionable content brief based on gap analysis"""        # Extract common themes from gap queries    all_words = " ".join(gap_queries).lower().split()    word_freq = {}    for word in all_words:        if len(word) > 3:  # Filter short words            word_freq[word] = word_freq.get(word, 0) + 1        top_keywords = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:10]        # Analyze competitor response patterns    avg_length = sum(len(resp.split()) for resp in competitor_responses) / len(competitor_responses)        brief = {        "category": query_category,        "priority": "High" if len(gap_queries) > 50 else "Medium",        "target_keywords": [kw[0] for kw in top_keywords],        "recommended_length": f"{int(avg_length * 1.2)}-{int(avg_length * 1.5)} words",        "content_angle": f"Address {len(gap_queries)} customer questions in {query_category}",        "sample_queries": gap_queries[:5]    }        return brief# Apply brief generation to top content gapscontent_briefs = []for row in content_gaps.take(10):  # Top 10 content opportunities    brief = generate_content_brief(        row["query_category"],        row["gap_queries"],        []  # Would include competitor responses in full implementation    )    content_briefs.append(brief)# Export content briefsimport jsonwith open("/tmp/content_briefs.json", "w") as f:    json.dump(content_briefs, f, indent=2)