rdd.foreachPartition {
    partitionOfRecords =>
      val connection = HdfsConnection.getHdfsConnection(config)
      partitionOfRecords.foreach(
        record => {
          // connection.writeUTF(record)
          connection.write(record.getBytes("UTF-8"))
          connection.writeBytes("\n")
        }
      )
      // 每次完了之后进行 flush
      try {
        connection.hflush()
      } catch {
        case e: Exception => logger.error(s"hflush exception: ${e.getMessage}")
      }
  }