mcpField
S
spark

spark

github.com/alphaonedev数据与数据库44 次安装
代码仓库

关于 spark

spark 是一项介绍 Apache Spark 的技能，用于分布式处理大型数据集，面向需要在集群上处理批次、流式、机器学习或图数据工作负载的数据工程师和分析人员。

主要功能

  1. 01基于 RDD 的分布式处理
  2. 02用于迭代任务的内存缓存
  3. 03支持 Scala、Python、Java 和 R
  4. 04提供 SQL、流处理、机器学习和图计算库
  5. 05集群执行与资源分配

适用场景

  1. 01对大型数据集运行 ETL 作业
  2. 02使用 Structured Streaming 分析流数据
  3. 03在集群上训练机器学习工作负载
  4. 04使用 GraphX 处理图数据

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/alphaonedev/openclaw-graph --skill spark

安装后在 Claude Code 中输入/spark确认已加载。

详细介绍

spark

用途

Apache Spark 是一个快速的分布式处理框架，利用内存计算处理大规模数据集。它支持在集群上高效地进行批处理、实时分析、机器学习和图处理。

适用场景

当数据集大于单台机器的内存时，可使用 Spark 进行处理，例如分析 TB 级日志数据或运行 ETL 作业。在需要快速迭代计算的场景中（例如机器学习算法），或需要与 Hadoop 等大数据生态系统集成时，也适合使用 Spark。对于 Pandas 等更简单的工具即可胜任的小规模任务，则不建议使用 Spark。

主要功能

  • 通过内存缓存加速迭代算法，例如使用 persist(StorageLevel.MEMORY_ONLY)。
  • 基于 RDD（弹性分布式数据集）实现容错分布式计算，并支持自动恢复。
  • 支持多种语言：Scala、Python、Java、R；例如，可使用 PySpark 和 from pyspark.sql import SparkSession 处理数据帧。
  • 内置库：用于结构化数据查询的 Spark SQL、用于机器学习的 MLlib、用于图处理的 GraphX，以及用于实时数据处理的 Structured Streaming。
  • 可扩展至数千个节点，并可通过 YARN 或 Kubernetes 动态分配资源。

使用模式

要使用 Spark 处理数据，请先在代码中创建 SparkSession。对于批处理作业，请通过 spark-submit 提交；对于交互式工作，请使用 Spark shell。请始终指定 master URL，例如集群模式下使用 "yarn"。读取文件或数据库中的数据源，使用 DataFrame 进行转换，并写出结果。对于流处理，请使用 Structured Streaming 实时处理 Kafka 主题。

示例 1：使用 PySpark 统计词频。

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WordCount").getOrCreate()
words = spark.read.text("hdfs://path/to/file.txt").rdd.flatMap(lambda x: x[0].split(" "))
counts = words.map(lambda x: (x, 1)).reduceByKey(lambda a, b: a + b)
counts.saveAsTextFile("hdfs://output/path")

示例 2：将 CSV 转换为 Parquet 的 ETL 作业。

spark = SparkSession.builder.master("local[*]").appName("ETL").getOrCreate()
df = spark.read.format("csv").option("header", "true").load("s3://bucket/data.csv")
df.write.format("parquet").mode("overwrite").save("hdfs://processed/data.parquet")

运行这些示例，请使用：spark-submit --master yarn --executor-memory 4g your_script.py。

常用命令/API

使用 spark-submit 运行应用程序：spark-submit --class MainClass --master yarn --deploy-mode cluster --driver-memory 2g your.jar arg1 arg2。要启动交互式会话，请运行 pyspark 或 spark-shell。常用 API 调用包括创建 SparkSession：SparkSession.builder().appName("App").master("local").getOrCreate()。使用 spark.read.csv("path", header=True, inferSchema=True) 读取数据。使用 DataFrame API 转换数据，例如 df.filter(df['age'] > 30).groupBy('department').count()。配置方面，请使用 SparkConf：conf = SparkConf().set("spark.executor.cores", "2")。设置用于访问集群的环境变量，例如用于主节点的 $SPARK_MASTER_URL。

集成说明

要将 Spark 与 Hadoop 集成，请将 $HADOOP_CONF_DIR 环境变量设置为 Hadoop 配置路径，然后将 YARN 用作 master。对于 Kafka，请在 spark-submit 中通过 --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2 添加连接器，并使用 spark.readStream.format("kafka").option("kafka.bootstrap.servers", "host:port").load() 读取流。使用 JDBC 连接数据库：df.write.jdbc(url="jdbc:postgresql://host/db", table="table", mode="append")，并确保类路径中包含 JDBC 驱动程序。可使用 spark-defaults.conf 等配置文件设置参数，例如 spark.sql.shuffle.partitions 200。

错误处理

遇到内存溢出错误时，可增加内存：在 spark-submit 中添加 --driver-memory 4g --executor-memory 8g。对于失败的任务，请查看 Spark UI http://driver-host:4040 中的日志，并使用 spark.task.maxFailures 配置设置重试次数上限。常见的序列化问题（例如 NotSerializableException）可通过使类支持序列化来解决，例如在 Java 中实现 Serializable。对于数据倾斜，可使用 df.repartition(100).write... 重新分区数据。请始终使用 try-except 包装 API 调用，例如 try: df = spark.read.csv("path") except Exception as e: print(e)。

图关系

关联：数据工程集群（例如，用于存储的 hadoop、用于编排的 airflow）。相关标签：big-data、distributed-computing。链接：与 kafka 集成以进行流处理，使用 hadoop 文件系统进行输入/输出。

评论

支持 Markdown