关于Fluentd转发日志至Spark及Spark数据位置与端口的技术咨询

Hi Harry，针对你的两个Spark相关问题，我给你整理了详细的解决方案和说明：

问题1：将Fluentd采集的日志转发至Spark #

你有几种可靠的方案可以实现这个需求，根据你的场景选择即可：

• 直接使用Fluentd Spark输出插件：社区提供了fluent-plugin-spark插件，能直接把Fluentd处理后的日志发送到Spark Streaming的接收器。

  1. 先安装插件：gem install fluent-plugin-spark
  2. 在Fluentd配置文件中添加输出规则：

  <match your.log.tag>
    @type spark
    host <spark-worker-or-driver-host>
    port <your-spark-streaming-listener-port>
    format json <!-- 或者你需要的日志格式 -->
  </match>
  

  3. 对应的Spark程序需要启动一个TCP接收器来接收数据（以Scala为例）：

  import org.apache.spark.streaming.{StreamingContext, Seconds}
  val ssc = new StreamingContext(sparkConf, Seconds(5))
  val logStream = ssc.socketTextStream("<fluentd-host-ip>", 9999) // 这里的端口要和Fluentd配置里的一致
  

• 借助Kafka做中间层（生产环境推荐）：如果日志量较大、需要系统解耦或者保证数据不丢失，更推荐用Kafka作为桥梁。

  1. Fluentd端安装Kafka插件并配置输出：

     <match your.log.tag>
       @type kafka
       brokers <kafka-broker-ip1>:9092,<kafka-broker-ip2>:9092
       default_topic spark_log_topic
       format json
     </match>
     

  2. Spark端用Structured Streaming读取Kafka数据：

     val df = spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "<kafka-broker-ip1>:9092")
       .option("subscribe", "spark_log_topic")
       .load()
     // 后续可以解析日志内容并做处理
     val logContent = df.selectExpr("CAST(value AS STRING)")
     

问题2：Spark的数据存储位置与运行端口说明 #

数据存储位置 #

• 临时中间数据：Spark运行作业时产生的临时数据，默认存在本地/tmp/spark-<随机后缀>目录下。你可以通过修改spark-defaults.conf中的spark.local.dir配置项，指定自定义路径：

  spark.local.dir /path/to/your/custom/temp/dir
  

• 持久化数据：如果你用cache()或persist()将RDD/DataFrame持久化，默认存储级别是内存优先（MEMORY_ONLY），内存不足时会自动溢出到spark.local.dir指定的磁盘路径。如果是持久化到外部存储（比如你已经配置的Cassandra），数据就存在Cassandra集群的对应位置。

常用运行端口 #

• Spark Master通信端口：默认7077，Worker节点通过这个端口注册到Master，可通过spark.master.port修改。
• Spark应用Web UI端口：默认4040，每个Spark应用都会启动独立的UI页面，端口会自动递增（4041、4042…）如果4040被占用，可通过spark.ui.port指定固定端口。
• Standalone模式Master Web UI：默认8080，用来查看集群节点状态，可通过spark.master.ui.port修改。
• Standalone模式Worker Web UI：默认8081，查看Worker节点的资源使用情况，可通过spark.worker.ui.port修改。
• Spark History Server端口：默认18080，用来查看已完成作业的历史日志，可通过spark.history.ui.port修改。

内容的提问来源于stack exchange，提问作者Coder