Uncategorized

The social media company inked three deals in the U.S. to power its data centers and offset its carbon footprint.

In this tutorial, we explore how to harness Apache Spark’s techniques using PySpark directly in Google Colab. We begin by setting up a local Spark session, then progressively move through transformations, SQL queries, joins, and window functions. We also build and evaluate a simple machine-learning model to predict user subscription types and finally demonstrate how to save and reload Parquet files. Also, we experience how Spark’s distributed data-processing capabilities can be leveraged for analytics and ML workflows even in a single-node Colab environment. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser !pip install -q pyspark==3.5.1 from pyspark.sql import SparkSession, functions as F, Window from pyspark.sql.types import IntegerType, StringType, StructType, StructField, FloatType from pyspark.ml.feature import StringIndexer, VectorAssembler from pyspark.ml.classification import LogisticRegression from pyspark.ml.evaluation import MulticlassClassificationEvaluator spark = (SparkSession.builder.appName(“ColabSparkAdvancedTutorial”) .master(“local[*]”) .config(“spark.sql.shuffle.partitions”, “4”) .getOrCreate()) print(“Spark version:”, spark.version) data = [ (1, “Alice”, “IN”, “2025-10-01”, 56000.0, “premium”), (2, “Bob”, “US”, “2025-10-03”, 43000.0, “standard”), (3, “Carlos”, “IN”, “2025-09-27”, 72000.0, “premium”), (4, “Diana”, “UK”, “2025-09-30”, 39000.0, “standard”), (5, “Esha”, “IN”, “2025-10-02”, 85000.0, “premium”), (6, “Farid”, “AE”, “2025-10-02”, 31000.0, “basic”), (7, “Gita”, “IN”, “2025-09-29”, 46000.0, “standard”), (8, “Hassan”, “PK”, “2025-10-01”, 52000.0, “premium”), ] schema = StructType([ StructField(“id”, IntegerType(), False), StructField(“name”, StringType(), True), StructField(“country”, StringType(), True), StructField(“signup_date”, StringType(), True), StructField(“income”, FloatType(), True), StructField(“plan”, StringType(), True), ]) df = spark.createDataFrame(data, schema) df.show() We begin by setting up PySpark, initializing the Spark session, and preparing our dataset. We create a structured DataFrame containing user information, including country, income, and plan type. This forms the foundation for all transformations and analyses that follow. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser df2 = (df.withColumn(“signup_ts”, F.to_timestamp(“signup_date”)) .withColumn(“year”, F.year(“signup_ts”)) .withColumn(“month”, F.month(“signup_ts”)) .withColumn(“is_india”, (F.col(“country”) == “IN”).cast(“int”))) df2.show() df2.createOrReplaceTempView(“users”) spark.sql(“”” SELECT country, COUNT(*) AS cnt, AVG(income) AS avg_income FROM users GROUP BY country ORDER BY cnt DESC “””).show() w = Window.partitionBy(“country”).orderBy(F.col(“income”).desc()) df_ranked = df2.withColumn(“income_rank_in_country”, F.rank().over(w)) df_ranked.show() def plan_priority(plan): if plan == “premium”: return 3 if plan == “standard”: return 2 if plan == “basic”: return 1 return 0 plan_priority_udf = F.udf(plan_priority, IntegerType()) df_udf = df_ranked.withColumn(“plan_priority”, plan_priority_udf(F.col(“plan”))) df_udf.show() We now perform various data transformations, add new columns, and register the DataFrame as a SQL table. We explore Spark SQL for aggregation and apply window functions to rank users by income. We also introduce a user-defined function (UDF) to assign priority levels to subscription plans. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser country_data = [ (“IN”, “Asia”, 1.42), (“US”, “North America”, 0.33), (“UK”, “Europe”, 0.07), (“AE”, “Asia”, 0.01), (“PK”, “Asia”, 0.24), ] country_schema = StructType([ StructField(“country”, StringType(), True), StructField(“region”, StringType(), True), StructField(“population_bn”, FloatType(), True), ]) country_df = spark.createDataFrame(country_data, country_schema) joined = df_udf.alias(“u”).join(country_df.alias(“c”), on=”country”, how=”left”) joined.show() region_stats = (joined.groupBy(“region”, “plan”) .agg(F.count(“*”).alias(“users”), F.round(F.avg(“income”), 2).alias(“avg_income”)) .orderBy(“region”, “plan”)) region_stats.show() We enrich our user dataset by joining it with country-level metadata that includes region and population. We then compute analytical summaries such as average income and user counts by region and plan type. This step demonstrates how Spark simplifies the seamless combination and aggregation of large datasets. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser ml_df = joined.withColumn(“label”, (F.col(“plan”) == “premium”).cast(“int”)).na.drop() country_indexer = StringIndexer(inputCol=”country”, outputCol=”country_idx”, handleInvalid=”keep”) country_fitted = country_indexer.fit(ml_df) ml_df2 = country_fitted.transform(ml_df) assembler = VectorAssembler(inputCols=[“income”, “country_idx”, “plan_priority”], outputCol=”features”) ml_final = assembler.transform(ml_df2) train_df, test_df = ml_final.randomSplit([0.7, 0.3], seed=42) lr = LogisticRegression(featuresCol=”features”, labelCol=”label”, maxIter=20) lr_model = lr.fit(train_df) preds = lr_model.transform(test_df) preds.select(“name”, “country”, “income”, “plan”, “label”, “prediction”, “probability”).show(truncate=False) evaluator = MulticlassClassificationEvaluator(labelCol=”label”, predictionCol=”prediction”, metricName=”accuracy”) acc = evaluator.evaluate(preds) print(“Classification accuracy:”, acc) We move into machine learning by preparing data for model training and feature engineering. We index categorical columns, assemble features, and train a logistic regression model to predict premium users. We then evaluate its accuracy, showcasing how Spark MLlib integrates easily into the data workflow. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser output_path = “/content/spark_users_parquet” joined.write.mode(“overwrite”).parquet(output_path) parquet_df = spark.read.parquet(output_path) print(“Parquet reloaded:”) parquet_df.show() recent = spark.sql(“”” SELECT name, country, income, signup_ts FROM users WHERE signup_ts >= ‘2025-10-01’ ORDER BY signup_ts DESC “””) recent.show() recent.explain() spark.stop() We conclude by writing the processed data to Parquet format and reading it back into Spark for verification. We run a SQL query to extract recent signups and inspect the query plan for optimization insights. Finally, we gracefully stop the Spark session to complete our workflow. In conclusion, we gain a practical understanding of how PySpark unifies data engineering and machine learning tasks within a single scalable framework. We witness how simple DataFrame transformations evolve into SQL analytics, feature engineering, and predictive modeling, all while staying within Google Colab. By experimenting with these concepts, we strengthen our ability to prototype and deploy Spark-based data solutions efficiently in both local and distributed setups. Check out the FULL CODES here. Feel free to check out our GitHub Page for Tutorials, Codes and Notebooks. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well. The post How to Build an End-to-End Data Engineering and Machine Learning Pipeline with Apache Spark and PySpark appeared first on MarkTechPost.

arXiv:2510.26271v1 Announce Type: new Abstract: Vision-language models (VLMs) exhibit uneven performance across languages, a problem that is often exacerbated when the model size is reduced. While Knowledge distillation (KD) demonstrates promising results in transferring knowledge from larger to smaller VLMs, applying KD in multilingualism is an underexplored area. This paper presents a controlled empirical study of KD behavior across five distillation approaches, isolating their effects on cross-lingual representation consistency and downstream performance stability under model compression. We study five distillation formulations across CLIP and SigLIP2, and evaluate them on in-domain retrieval and out-of-domain visual QA. We find that some configurations preserve or even improve multilingual retrieval robustness despite halving model size, but others fail to maintain cross-task stability, exposing design-sensitive trade-offs that aggregate accuracy alone does not reveal.

arXiv:2510.26802v1 Announce Type: cross Abstract: Recent video generation models can produce high-fidelity, temporally coherent videos, indicating that they may encode substantial world knowledge. Beyond realistic synthesis, they also exhibit emerging behaviors indicative of visual perception, modeling, and manipulation. Yet, an important question still remains: Are video models ready to serve as zero-shot reasoners in challenging visual reasoning scenarios? In this work, we conduct an empirical study to comprehensively investigate this question, focusing on the leading and popular Veo-3. We evaluate its reasoning behavior across 12 dimensions, including spatial, geometric, physical, temporal, and embodied logic, systematically characterizing both its strengths and failure modes. To standardize this study, we curate the evaluation data into MME-CoF, a compact benchmark that enables in-depth and thorough assessment of Chain-of-Frame (CoF) reasoning. Our findings reveal that while current video models demonstrate promising reasoning patterns on short-horizon spatial coherence, fine-grained grounding, and locally consistent dynamics, they remain limited in long-horizon causal reasoning, strict geometric constraints, and abstract logic. Overall, they are not yet reliable as standalone zero-shot reasoners, but exhibit encouraging signs as complementary visual engines alongside dedicated reasoning models. Project page: https://video-cof.github.io

In this tutorial, we build an Agentic Data and Infrastructure Strategy system using the lightweight Qwen2.5-0.5B-Instruct model for efficient execution. We begin by creating a flexible LLM agent framework and then develop specialized agents that handle different layers of data management, from ingestion and quality analysis to infrastructure optimization. We integrate these agents into an orchestrator that coordinates their interactions, ensuring smooth multi-agent collaboration across the data pipeline. Through hands-on examples like e-commerce and IoT pipelines, we explore how autonomous decision-making can streamline complex data operations. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser !pip install -q transformers torch accelerate datasets huggingface_hub import torch from transformers import AutoModelForCausalLM, AutoTokenizer import json, time from typing import List, Dict, Any from dataclasses import dataclass from datetime import datetime import pandas as pd class LightweightLLMAgent: def __init__(self, role: str, model_name: str = “Qwen/Qwen2.5-0.5B-Instruct”): self.role = role self.model_name = model_name self.device = “cuda” if torch.cuda.is_available() else “cpu” print(f”Loading {model_name} for {role} agent on {self.device}…”) self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16 if self.device == “cuda” else torch.float32, device_map=”auto” ) self.conversation_history = [] def generate_response(self, prompt: str, max_tokens: int = 150) -> str: messages = [ {“role”: “system”, “content”: f”You are a {self.role} agent in a data infrastructure system.”}, {“role”: “user”, “content”: prompt} ] text = self.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) model_inputs = self.tokenizer([text], return_tensors=”pt”).to(self.device) with torch.no_grad(): generated_ids = self.model.generate( model_inputs.input_ids, max_new_tokens=max_tokens, temperature=0.7, do_sample=True, top_p=0.95 ) generated_ids = [output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)] response = self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0] self.conversation_history.append({“prompt”: prompt, “response”: response}) return response We start by setting up the lightweight LLM agent infrastructure using the Qwen2.5-0.5B-Instruct model. We load the model and tokenizer, and define a base agent class capable of handling contextual conversations and generating intelligent responses. This forms the core foundation upon which our specialized agents operate efficiently within Colab. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser class DataIngestionAgent(LightweightLLMAgent): def __init__(self): super().__init__(role=”Data Ingestion Specialist”) def analyze_data_source(self, source_info: Dict) -> Dict: prompt = f”””Analyze this data source and provide ingestion strategy: Source Type: {source_info.get(‘type’, ‘unknown’)} Volume: {source_info.get(‘volume’, ‘unknown’)} Frequency: {source_info.get(‘frequency’, ‘unknown’)} Provide a brief strategy focusing on: 1) Ingestion method, 2) Key considerations.””” strategy = self.generate_response(prompt, max_tokens=100) return {“source”: source_info, “strategy”: strategy, “timestamp”: datetime.now().isoformat()} class DataQualityAgent(LightweightLLMAgent): def __init__(self): super().__init__(role=”Data Quality Analyst”) def assess_data_quality(self, data_sample: Dict) -> Dict: prompt = f”””Assess data quality for this sample: Completeness: {data_sample.get(‘completeness’, ‘N/A’)}% Consistency: {data_sample.get(‘consistency’, ‘N/A’)}% Issues Found: {data_sample.get(‘issues’, 0)} Provide brief quality assessment and top 2 recommendations.””” assessment = self.generate_response(prompt, max_tokens=100) return {“assessment”: assessment, “severity”: self._calculate_severity(data_sample), “timestamp”: datetime.now().isoformat()} def _calculate_severity(self, data_sample: Dict) -> str: completeness = data_sample.get(‘completeness’, 100) consistency = data_sample.get(‘consistency’, 100) avg_score = (completeness + consistency) / 2 if avg_score >= 90: return “LOW” elif avg_score >= 70: return “MEDIUM” else: return “HIGH” We design the Data Ingestion and Data Quality agents to focus on structured analysis of data pipelines. We let the ingestion agent determine the best approach to data flow, while the quality agent evaluates data completeness, consistency, and issues to provide actionable insights. Together, they establish the first two layers of autonomous data management. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser class InfrastructureOptimizationAgent(LightweightLLMAgent): def __init__(self): super().__init__(role=”Infrastructure Optimization Specialist”) def optimize_resources(self, metrics: Dict) -> Dict: prompt = f”””Analyze infrastructure metrics and suggest optimizations: CPU Usage: {metrics.get(‘cpu_usage’, 0)}% Memory Usage: {metrics.get(‘memory_usage’, 0)}% Storage: {metrics.get(‘storage_used’, 0)}GB / {metrics.get(‘storage_total’, 0)}GB Query Latency: {metrics.get(‘query_latency’, 0)}ms Provide 2 optimization recommendations.””” recommendations = self.generate_response(prompt, max_tokens=100) return {“current_metrics”: metrics, “recommendations”: recommendations, “priority”: self._calculate_priority(metrics), “timestamp”: datetime.now().isoformat()} def _calculate_priority(self, metrics: Dict) -> str: cpu = metrics.get(‘cpu_usage’, 0) memory = metrics.get(‘memory_usage’, 0) if cpu > 85 or memory > 85: return “CRITICAL” elif cpu > 70 or memory > 70: return “HIGH” else: return “NORMAL” We develop the Infrastructure Optimization Agent to continuously analyze key metrics like CPU, memory, and storage utilization. We use it to generate intelligent optimization suggestions, helping us maintain high performance and resource efficiency. This agent ensures that our infrastructure remains responsive and scalable during data operations. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser class AgenticDataOrchestrator: def __init__(self): print(“n” + “=”*70) print(“Initializing Agentic Data Infrastructure System”) print(“=”*70 + “n”) self.ingestion_agent = DataIngestionAgent() self.quality_agent = DataQualityAgent() self.optimization_agent = InfrastructureOptimizationAgent() self.execution_log = [] def process_data_pipeline(self, pipeline_config: Dict) -> Dict: results = {“pipeline_id”: pipeline_config.get(“id”, “unknown”), “start_time”: datetime.now().isoformat(), “stages”: []} print(“n[Stage 1] Data Ingestion Analysis”) ingestion_result = self.ingestion_agent.analyze_data_source(pipeline_config.get(“source”, {})) print(f”Strategy: {ingestion_result[‘strategy’][:150]}…”) results[“stages”].append({“stage”: “ingestion”, “result”: ingestion_result}) print(“n[Stage 2] Data Quality Assessment”) quality_result = self.quality_agent.assess_data_quality(pipeline_config.get(“quality_metrics”, {})) print(f”Assessment: {quality_result[‘assessment’][:150]}…”) print(f”Severity: {quality_result[‘severity’]}”) results[“stages”].append({“stage”: “quality”, “result”: quality_result}) print(“n[Stage 3] Infrastructure Optimization”) optimization_result = self.optimization_agent.optimize_resources(pipeline_config.get(“infrastructure_metrics”, {})) print(f”Recommendations: {optimization_result[‘recommendations’][:150]}…”) print(f”Priority: {optimization_result[‘priority’]}”) results[“stages”].append({“stage”: “optimization”, “result”: optimization_result}) results[“end_time”] = datetime.now().isoformat() results[“status”] = “completed” self.execution_log.append(results) return results def generate_summary_report(self) -> pd.DataFrame: if not self.execution_log: return pd.DataFrame() summary_data = [] for log in self.execution_log: summary_data.append({“Pipeline ID”: log[“pipeline_id”], “Start Time”: log[“start_time”], “Status”: log[“status”], “Stages Completed”: len(log[“stages”])}) return pd.DataFrame(summary_data) We built an Agentic Data Orchestrator to coordinate all specialized agents under a unified workflow. We use it to manage end-to-end pipeline execution, triggering ingestion, quality checks, and optimization sequentially. By doing this, we bring structure, collaboration, and automation to the entire multi-agent system. Check out the FULL CODES here. Copy CodeCopiedUse a different Browser def main(): orchestrator = AgenticDataOrchestrator() print(“n” + “=”*70) print(“EXAMPLE 1: E-commerce Data Pipeline”) print(“=”*70) ecommerce_pipeline = { “id”: “ecommerce_pipeline_001”, “source”: {“type”: “REST API”, “volume”: “10GB/day”, “frequency”: “real-time”}, “quality_metrics”: {“completeness”: 87, “consistency”: 92, “issues”: 15}, “infrastructure_metrics”: {“cpu_usage”: 78, “memory_usage”: 82, “storage_used”: 450, “storage_total”: 1000, “query_latency”: 250} } result1 = orchestrator.process_data_pipeline(ecommerce_pipeline) print(“nn” + “=”*70) print(“EXAMPLE 2: IoT Sensor Data Pipeline”) print(“=”*70) iot_pipeline = { “id”: “iot_pipeline_002”, “source”: {“type”: “Message Queue (Kafka)”, “volume”: “50GB/day”, “frequency”: “streaming”}, “quality_metrics”: {“completeness”: 95, “consistency”: 88, “issues”: 8}, “infrastructure_metrics”: {“cpu_usage”: 65, “memory_usage”: 71, “storage_used”: 780, “storage_total”: 2000, “query_latency”: 180} } result2 = orchestrator.process_data_pipeline(iot_pipeline) print(“nn” + “=”*70) print(“EXECUTION SUMMARY REPORT”) print(“=”*70 + “n”) summary_df = orchestrator.generate_summary_report() print(summary_df.to_string(index=False)) print(“n” + “=”*70) print(“Tutorial Complete!”) print(“=”*70) print(“nKey Concepts Demonstrated:”) print(“✓ Lightweight LLM agent architecture”) print(“✓ Specialized agents for different data tasks”) print(“✓ Multi-agent orchestration”) print(“✓ Infrastructure monitoring and optimization”) print(“✓ Autonomous decision-making in data pipelines”) if __name__ == “__main__”: main() We demonstrate our complete

arXiv:2509.12760v2 Announce Type: replace-cross Abstract: We introduce the Similarity-Distance-Magnitude (SDM) activation function, a more robust and interpretable formulation of the standard softmax activation function, adding Similarity (i.e., correctly predicted depth-matches into training) awareness and Distance-to-training-distribution awareness to the existing output Magnitude (i.e., decision-boundary) awareness, and enabling interpretability-by-exemplar via dense matching. We further introduce the SDM estimator, based on a data-driven partitioning of the class-wise empirical CDFs via the SDM activation, to control the class- and prediction-conditional accuracy among selective classifications. When used as the final-layer activation over pre-trained language models for selective classification, the SDM estimator is more robust to co-variate shifts and out-of-distribution inputs than existing calibration methods using softmax activations, while remaining informative over in-distribution data.

arXiv:2510.04226v4 Announce Type: replace Abstract: Large language models (LLMs) tend to generate lexically, semantically, and stylistically homogenous texts. This poses a risk of knowledge collapse, where homogenous LLMs mediate a shrinking in the range of accessible information over time. Existing works on homogenization are limited by a focus on closed-ended multiple-choice setups or fuzzy semantic features, and do not look at trends across time and cultural contexts. To overcome this, we present a new methodology to measure epistemic diversity, i.e., variation in real-world claims in LLM outputs, which we use to perform a broad empirical study of LLM knowledge collapse. We test 27 LLMs, 155 topics covering 12 countries, and 200 prompt variations sourced from real user chats. For the topics in our study, we show that while newer models tend to generate more diverse claims, nearly all models are less epistemically diverse than a basic web search. We find that model size has a negative impact on epistemic diversity, while retrieval-augmented generation (RAG) has a positive impact, though the improvement from RAG varies by the cultural context. Finally, compared to a traditional knowledge source (Wikipedia), we find that country-specific claims reflect the English language more than the local one, highlighting a gap in epistemic representation

arXiv:2510.24789v1 Announce Type: new Abstract: Watermarking has been proposed as a lightweight mechanism to identify AI-generated text, with schemes typically relying on perturbations to token distributions. While prior work shows that paraphrasing can weaken such signals, these attacks remain partially detectable or degrade text quality. We demonstrate that cross-lingual summarization attacks (CLSA) — translation to a pivot language followed by summarization and optional back-translation — constitute a qualitatively stronger attack vector. By forcing a semantic bottleneck across languages, CLSA systematically destroys token-level statistical biases while preserving semantic fidelity. In experiments across multiple watermarking schemes (KGW, SIR, XSIR, Unigram) and five languages (Amharic, Chinese, Hindi, Spanish, Swahili), we show that CLSA reduces watermark detection accuracy more effectively than monolingual paraphrase at similar quality levels. Our results highlight an underexplored vulnerability that challenges the practicality of watermarking for provenance or regulation. We argue that robust provenance solutions must move beyond distributional watermarking and incorporate cryptographic or model-attestation approaches. On 300 held-out samples per language, CLSA consistently drives detection toward chance while preserving task utility. Concretely, for XSIR (explicitly designed for cross-lingual robustness), AUROC with paraphrasing is $0.827$, with Cross-Lingual Watermark Removal Attacks (CWRA) [He et al., 2024] using Chinese as the pivot, it is $0.823$, whereas CLSA drives it down to $0.53$ (near chance). Results highlight a practical, low-cost removal pathway that crosses languages and compresses content without visible artifacts.