<match *.**>
  type copy
  <store>
    type secure_forward
    shared_key FLUENTD_SECRET
    self_hostname node1.example.com
    secure true
    ca_cert_path /path/to/certificate/ca_cert.pem
    <server>
      host **.**.***.***
    </server>
  </store>
  <store>
    type kinesis
    stream_name *******
    aws_key_id **************
    aws_sec_key ******************
    region ap-northeast-1
    random_partition_key true
  </store>
</match>

libraryDependencies += "org.apache.spark" % "spark-streaming-kinesis-asl_2.10" % "1.5.0

object wordCount{
  def main(args: Array[String]) {
    val accessKey = "**************
    val secretKey = "******************
    //val credentials = new DefaultAWSCredentialsProviderChain().getCredentials()
    val credentials = new BasicAWSCredentials(accessKey, secretKey)
    val kinesisClient = new AmazonKinesisClient(credentials)
    val streamName = "********
    val endpointUrl = "https://kinesis.ap-northeast-1.amazonaws.com"
    kinesisClient.setEndpoint(endpointUrl)
    val numShards = kinesisClient.describeStream(streamName).getStreamDescription().getShards().size

    // In this example, we're going to create 1 Kinesis Receiver/input DStream for each shard.
    // This is not a necessity; if there are less receivers/DStreams than the number of shards,
    // then the shards will be automatically distributed among the receivers and each receiver
    // will receive data from multiple shards.
    val numStreams = numShards

    // Spark Streaming batch interval
    val batchInterval = Seconds(10)

    // Kinesis checkpoint interval is the interval at which the DynamoDB is updated with information
    // on sequence number of records that have been received. Same as batchInterval for this
    // example.
    val kinesisCheckpointInterval = batchInterval

    // Get the region name from the endpoint URL to save Kinesis Client Library metadata in
    // DynamoDB of the same region as the Kinesis stream
    val regionName = RegionUtils.getRegionByEndpoint(endpointUrl).getName()

    // Setup the SparkConfig and StreamingContext
    val sparkConfig = new SparkConf().setAppName("KinesisWordCountASL")
    val ssc = new StreamingContext(sparkConfig, batchInterval)

    // Create the Kinesis DStreams
    val kinesisStreams = (0 until numStreams).map { i =>
      KinesisUtils.createStream(ssc, "KinesisWordCountASL", streamName, endpointUrl, regionName,
        InitialPositionInStream.LATEST, kinesisCheckpointInterval, StorageLevel.MEMORY_AND_DISK_2,
        accessKey, secretKey)
    }


    // Union all the streams
    val unionStreams = ssc.union(kinesisStreams)

    // Convert each line of Array[Byte] to String, and split into words
    val words = unionStreams.flatMap(byteArray => new String(byteArray).split(" "))

    // Map each word to a (word, 1) tuple so we can reduce by key to count the words
    val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)

    val hadoopConf = ssc.sparkContext.hadoopConfiguration
    hadoopConf.set("fs.s3n.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
    hadoopConf.set("fs.s3n.awsAccessKeyId", "********")
    hadoopConf.set("fs.s3n.awsSecretAccessKey", "******************")
    wordCounts.saveAsTextFiles("s3n://emr-datacollect/output/test/")

    // Print the first 10 wordCounts
    wordCounts.print()

    // Start the streaming context and await termination
    ssc.start()
    ssc.awaitTermination()

  }
}