Spark利用ConfigFactory获取配置文件详解

获取配置文件中的配置

在实际开发中，项目中的一些配置文件会放在classpath下的resources文件中，然后在程序中会加载配置文件。

加载配置文件两种方式：

利用类加载器。
利用Typesafe的Config库（ConfigFactory）实现。

实现方式一：利用类加载器

    private static final Properties props;

    static {
        props = new Properties();
        try {
            props.load(ConfigUtil.class.getResourceAsStream("/profile.properties"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    // 获取配置
    props.getProperty(key)

实现方式二：利用Typesafe的Config库

Typesafe的Config库，是代码精简、功能灵活、API友好。它也是Akka的配置管理库。

利用config库下的ConfigFactory类来加载配置文件。

所需jar包

      <dependency>
        <groupId>com.typesafe</groupId>
        <artifactId>config</artifactId>
        <version>1.3.3</version>
      </dependency>
      
    <!-- akka的actor依赖 -->
    <dependency>
        <groupId>com.typesafe.akka</groupId>
        <artifactId>akka-actor_2.11</artifactId>
        <version>2.4.17</version>
    </dependency>

两个jar包的区别：
第一个jar包是第二个jar包的依赖jar包。

ConfigFactory实现

ConfigFactory.load() 默认加载classpath下的配置文件：
加载顺序为：application.conf ---> application.json ----> application.properties

package com.ly.jtbi

import com.typesafe.config.ConfigFactory


/**
  * @Auther: fc.w
  * @Date: 2019/3/14
  */
object CommonUtils {

  val config = ConfigFactory.load()
  val env  = config.getString("env")

  val KAFKA_BOOTSTRAP_SERVERS_KEY   = config.getString("kafka.bootstrap.servers.key")
  val KAFKA_BOOTSTRAP_SERVERS_VALUE = config.getString("kafka.bootstrap.servers.value")
  val KAFKA_GROUP_ID_KEY   = config.getString("kafka.group.id.key")
  val KAFKA_GROUP_ID_VALUE = config.getString("kafka.group.id.value")
  val KAFKA_TOPIC = config.getString("kafka.topic")

}

各种配置文件实现

其中，application.properties中的配置文件格式如下：

综合案例

/**
  * @Auther: fc.w
  * @Date: 2019/3/13
  */
object App {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)            // 默认 ProcessingTime MW默认间隔0ms； 设置为EventTime 或 IngestionTime MW默认间隔200ms
    env.enableCheckpointing(1000)                                      // 默认 500毫秒
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) // 默认 Exactly_once

    val kafkaProps = new Properties()
    kafkaProps.setProperty(CommonUtils.KAFKA_BOOTSTRAP_SERVERS_KEY, CommonUtils.KAFKA_BOOTSTRAP_SERVERS_VALUE)
    kafkaProps.setProperty(CommonUtils.KAFKA_GROUP_ID_KEY, CommonUtils.KAFKA_GROUP_ID_VALUE)
  }
}