# 上传一个文件 -f表示覆盖
hadoop fs -put -f jour.txt /user/dahu/jour/
# 下载
hadoop fs -get /user/dahu/jour/jour.txt
# ls
hadoop fs -ls /user/dahu/
# 删除 -r表示递归,删除目录
hadoop fs -rm /user/dahu/jour/jour.txt
hadoop fs -rm -r /user/dahu/jour

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "use_a_separate_group_id_for_each_stream",
  "auto.offset.reset" -> "latest",
  "enable.auto.commit" -> (false: java.lang.Boolean)
)

val topics = Array("topicA", "topicB")
val stream = KafkaUtils.createDirectStream[String, String](
  streamingContext,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

stream.map(record => (record.key, record.value))

dstream.foreachRDD { rdd =>
  val connection = createNewConnection()  // executed at the driver
  rdd.foreach { record =>
    connection.send(record) // executed at the worker
  }
}
// 上面的写法会导致 connection 不可序列化的错误：Task not serializable
// RDD 的函数 (map,foreach) 会被序列化发送到 worker 节点执行，但是 connection 是和 tcp 连接，和机器绑定的，无法序列化

dstream.foreachRDD { rdd =>
  rdd.foreach { record =>  // on worker node
    val connection = createNewConnection() // 给每个 record 处理时新建一个连接，会导致严重的数据库连接性能问题
    connection.send(record)
    connection.close()
  }
}

// 更好的方式是给每个 partation 新建一个连接
dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val connection = createNewConnection() 
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
  }
}

// 最好的方法是维护一个静态线程池：
[ConnectionPool](https://github.com/RedisLabs/spark-redis/blob/master/src/main/scala/com/redislabs/provider/redis/ConnectionPool.scala)
// then use in partition
dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}
// Note that the connections in the pool should be lazily created on demand and timed out if not used for a while. 
// This achieves the most efficient sending of data to external systems.

// 示例
case class RedisCluster(clusterHosts: String, password: String) extends Serializable {

  def this(conf: SparkConf) {
    this(
      conf.get("spark.redis.host", Protocol.DEFAULT_HOST),
      conf.get("spark.redis.auth", null)
    )
  }

  /**
   *
   * @return use for JedisCluster or JedisPool
   */
  def toSet(): java.util.Set[HostAndPort] = {
    val nodes: mutable.Set[HostAndPort] = mutable.Set()
    for (host_port <- clusterHosts.split(",")) {
      val hp = host_port
      print(hp)
      nodes += HostAndPort.from(host_port)
    }
    nodes.asJava
  }

}

object RedisClusterUtils extends Serializable {

  @transient private lazy val pools: ConcurrentHashMap[RedisCluster, JedisCluster] =
    new ConcurrentHashMap[RedisCluster, JedisCluster]()

  /**
   * 获取一个 JedisCluster
   * @param rc
   * @return
   */
  def connect(rc: RedisCluster): JedisCluster = {

    pools.getOrElseUpdate(rc, {
      val poolConfig = new JedisPoolConfig();
      poolConfig.setMaxTotal(250)
      poolConfig.setMaxIdle(32)
      poolConfig.setTestOnBorrow(false)
      poolConfig.setTestOnReturn(false)
      poolConfig.setTestWhileIdle(false)
      poolConfig.setNumTestsPerEvictionRun(-1)

      val jedisCluster = new JedisCluster(rc.toSet(),
        3000,
        3000,
        5,
        rc.password,
        poolConfig)

      jedisCluster
    })
  }

  /**
   * 查询币种对应汇率
   * @param jedisCluster 目标 redis
   * @param ccyCd 币种代码
   * @return 折美元汇率
   */
  def getCcyRatio(jedisCluster: JedisCluster, ccyCd:String): Double ={
    val res = jedisCluster.get("CCY:"+ccyCd)
    res.split(":")(2).toDouble
  }
}