AWS Kinesis Stream

AWS Kinesis Stream是近乎实时收集和处理大数据记录流。可以理解为近乎实时数据传输管道，相当于以前用FTP传输数据，但它是用API且将数据作为字节流传输。1秒的平均传播延迟。

使用场景

用于快速而持续的数据引入和聚合
使用的数据类型可以包括 IT 基础设施日志数据、应用程序日志、社交媒体、市场数据源和 Web点击流数据。由于数据引入和处理的响应时间是实时的，因此处理通常是轻量级的。

日志引入:推送系统和应用程序日志=metrics
实时数据分析:实时处理网站点击流
实时聚合数据，然后将聚合数据加载到数据仓库或 map-reduce群集

架构

关键在于分片shard，流可以指定多个分片，而AWS Kinesis需要花时间为该流分配出分片。分为生产者和使用者，分别是产生数据和消费数据。

Kinesis Data Stream:

保留期(retention period)是数据记录在添加到流中后可供访问的时间长度。在创建之后，流的保留期设置
为默认值 24小时。最多到7天。

分片是流中数据记录的唯一标识序列。一个流由一个或多个分片组成，每个分片提供一个固定的容量单
位。流的总容量是其分片容量的总和。
分区键用于按分片对流中的数据进行分组。MD5 哈希函数用于将分区键映射到 128 位整数值并将关联的
数据记录映射到分片。应用程序需要对每个数据指定分区键。在创建流时，您将指定流的分片数。

number_of_shards = max(incoming_write_bandwidth_in_KiB/1024, outgoing_read_bandwidth_in_KiB/2048)

使用者称为Amazon Kinesis Data Streams Application

Kinesis Client Library(供使用者使用，是Kinesis服务的客户端)KCL 负责许多与分布式计算相关的复杂
任务，例如对多个实例进行负载均衡、
对实例故障做出响应、对已处理的数据执行检查点操作和对重新分 片做出反应。KCL 可让您将精力
放在编写记录处理逻辑上。
它使用Amazon DynamoDB表存储控制数据。它会为每个正在处理数据的应用程序创建一个表。实例中，
选择 KinesisDataVisSampleApp-KCLDynamoDBTable-[randomString] 表。
在表中有两个条目，指示特 定分片 (leaseKey)、流中的位置 (checkpoint) 和读取数据的应用程序 
(leaseOwner)。

用 AWS KMS 主密钥进行加密

最佳实践

AWS实例是统计URL实时数目(2s window)。

分析实时股票数据

底层API

创建和描述stream，需要的时间创建stream主要用于分配shard

create-stream --stream-name Foo --shard-count 1

describe-stream --stream-name Foo

Response:

{

   "StreamDescription": {

       "StreamStatus": "ACTIVE",

       "StreamName": "Foo",

       "StreamARN": "arn:aws:kinesis:us-west-2:account-id:stream/Foo",

       "Shards": [

           {

               "ShardId": "shardId-000000000000",

               "HashKeyRange": {

                   "EndingHashKey": "170141183460469231731687303715884105727",

                   "StartingHashKey": "0"

               },

               "SequenceNumberRange": {

                   "StartingSequenceNumber": "495469866831355442865074579357546397794"

               }

           }

       ]

   }

}

AWS 区域的默认分片限制为 500 分片

单个分片可以提取多达每秒 1 MiB 的数据 (包括分区键) 或每秒写入 1000 个记录.

每个 PutRecord 调用都需要流名称、分区键和创建者正在添加到流的数据记录。

put-record --stream-name Foo --partition-key 123 --data testdata

Response:

{ "ShardId": "shardId-000000000000", "SequenceNumber": "49546156785154" }

要将数据放入流，您必须指定流的名称、分区键和要添加到流的数据 Blob。GetRecords
可以从单个分片中检索多达每个调用 10 MiB 数据，每个调用多达 10000 个记录。每个分片可以通过 GetRecords 支持每秒 2 MiB 的最大总数据读取速率。

首先获得分片的迭代器（指针）

get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name Foo

得到迭代器的id，然后将其带入到getRecords API

get-records --shard-iterator AAAAAAAAAAHSywlj

Response:

{ "Records":[ { "Data":"dGVzdGRhdGE=", "PartitionKey":"123”, "ApproximateArrivalTimestamp": 
1.441215410867E9, 
"SequenceNumber":"49544985256907370027570885864065577703022652638596431874" } ], 
"MillisBehindLatest":24000, "NextShardIterator":"AAAA

Data为base64编码

NextShardIterator下一个iterator，即使流没有新数据

当Records为空的时候才表示全部读完

MillisBehindLatest表示距流末端（最新数据）还有多久。零值指示正进行记录处理，此时没有新
的记录要处理

术语检查点操作是指记录到流中目前已使用和处理的数据记录所在的点，这样一来，当应用程序
发生崩溃时，系统将从该点读取流，而不是从头开始读取流。
只要记录SequenceNumber。然后获取Iterator时指定这个checkpoint即可。
GetShardIterator::StartingSequenceNumber

封装层

生产器  - KPL

Kinesis Producer Library (KPL) 简化了创建器应用程序的开发.KPL 是一个易于使用的、高度可配置的库，可帮助您对 Kinesis data stream 进行写入. 由于 KPL 可在将记录发送到 Kinesis Data Streams 之前对其进行缓冲处理，从而产生高吞入量。KPL 会导致库（用户可配置的）中产生高达 RecordMaxBufferedTime 的额外处理延 迟。RecordMaxBufferedTime 值越大，产生的包装效率和性能就越高。

使用器 - KCL
RecordMaxBufferedTime 上述以提到，这是应用程序可以用到的库

使用期 - Kinesis Firehose - 更高级封装

Amazon Kinesis Data Firehose，AWS Lambda 开发使用器，Amazon Kinesis Data Analytics 开发
使用器。您可以使用 Kinesis Data Firehose 读取和处理 Kinesis 流中的记录。Kinesis Data Firehose 
是一个完全托管 的服务，用于将实时流数据传输到目标

（如 Amazon S3、Amazon Redshift、Amazon Elasticsearch Service 和 Splunk）。也是不再需要
编写Java代码就可以直接将数据加载到目标。

例如目标为redshift

{"TICKER_SYMBOL":"QXZ","SECTOR":"HEALTHCARE","CHANGE":-0.05,"PRICE":84.51}

=>

create table firehose_test_table

(

TICKER_SYMBOL varchar(4),

SECTOR varchar(16),

CHANGE float,

PRICE float

);

Firehose还可以单独使用，用于让数据流传输到适用的目标如S3。与SQS非常相似也是提供一个有时限的缓存，不同系统之间传输信息，区别在于实时vs非实时，大数据vs msg。
与Kinesis区别是firehose缓存大小对于S3只有128M，比Kinesis小很大，缓存的时间buffer interval也只有60s到900s，比kenesis小很多，目标也只能有限几个AWS产品，而kinesis可以是任何下游产品。kinesis类似于Kafka。
最佳实践为music应用，metrics传输到第三方partner，DDB->lambda->firehose->S3->Redshift

功能

单个分片->单个工作线程取数据。如果数据量大，可以将某几个shard聚合再放入到下一层的
stream。通过两层stream聚合来更快处理数据。

支持重新分片，这使您能够调整流中的分片数量以适应流中数据流量的变化。

增强型扇出功能 是一种 Amazon Kinesis Data Streams 功能，使用者利用此功能能够接收数据流（其
中每分 片每秒的专用吞吐量高达 2 MiB 数据）中的记录。此吞吐量是专用的，
这意味着，使用增强型扇出 功能的使用者不必与接收流中数据的其他使用者争夺。
Kinesis Data Streams 将流中的数据记录推送到使用 增强型扇出功能的使用者。
因此，这些使用者无需轮询数据。

Amazon Kinesis Data Streams 维度与指标用cloudwatch来记录

常见问题

（读取数据终止条件）即使流中有数据，GetRecords 仍然返回空记录阵列

ShardIterator 指向的分片部分附近没有数据。 此情况很微妙，但却是避免在检索数据时搜寻时间
无止境（延迟）的一种必要的设计折衷。因此，流使 用应用程序应循环并调用 GetRecords，并
且理所当然地处理空记录。 在生产场景中，仅当 NextShardIterator 值为 NULL 时，
才应退出连续循环。当 NextShardIterator 为 NULL 时，这意味着当前分片已关闭，ShardIterator
值的指向应越过最后一条记录。如果使用应用 程序从不调用 SplitShard 或 MergeShards，分片将
保持打开状态，并且对GetRecords 的调用从不返回为 NextShardIterator 的 NULL 值。

Java代码示例

下面发布者通过kinesisClient发布一个股票交易到AWS。而流事件是生产者自定义的。这是流事件的ID，通过这个ID，使用者者可只关注某些流事件。
private static void sendStockTrade(StockTrade trade, AmazonKinesis kinesisClient, String streamName) {
    byte[] bytes = trade.toJsonAsBytes();

    PutRecordRequest putRecord = new PutRecordRequest();
    putRecord.setStreamName(streamName);
 
    // We use the ticker symbol as the partition key, explained in the Supplemental Information section     //below.
    putRecord.setPartitionKey(trade.getTickerSymbol());
    putRecord.setData(ByteBuffer.wrap(bytes));

    kinesisClient.putRecord(putRecord);
}

使用者：

订阅者通过kinesisClient得到record，然后可以得到trade
StockTrade trade = StockTrade.fromJsonAsBytes(record.getData().array());
stockStats.addStockTrade(trade);

与SQS的区别：

Kinesis是实时的且可用于大数据，而SQS是非实时处理任务，将同步变异步解耦削峰。


ref:
官方指南
FireHose官方指南
分布式发布订阅系统
概念
发布者
订阅者
访问权限
与SQS区别

JSON和POJO转换 -- fasterxml

本篇介绍fasterxml的json库，可以用于POJO互换Json payload。类似的库有.Net的Newtonsoft。fasterxml用到两个lib：jackson-databind和jackson-core。

POJO不能含有构造器。下面代码首先用Customer的实例产生json，再读入json转换成cusomer实例。以下是json结果：
{"customerId":555,"address":{"city":"Seattle","state":null}}

MyJsonHelper.java:
public class MyJsonHelper {

public void toObject() throws JsonParseException, JsonMappingException, IOException {
Customer customer = mapper.readValue(new File("test.json"), Customer.class);
System.out.println(customer.getAddress());
}

public void toJson() throws JsonGenerationException, JsonMappingException, IOException {

Customer customer = new Customer();
Address address = new Address();
address.setCity("Seattle");
customer.setCustomerId(555);
customer.setAddress(address);

//Object to JSON in file
mapper.writeValue(new File("test.json"), customer);

//Object to JSON in String
String jsonInString = mapper.writeValueAsString(customer);
System.out.println(jsonInString);
}

private ObjectMapper mapper = new ObjectMapper();
}

Customer.java:
public class Customer {

private int customerId = 4;
private Address address;

public void setCustomerId(int custid){
this.customerId = custid;
}

public int getCustomerId(){
return customerId;
}

public void setAddress(Address address){
this.address = address;
}

public Address getAddress(){
return address;
}
}

Address.java:
public class Address {

private String city;
private String state;

public void setCity(String city){
this.city = city;
}

public String getCity(){
return city;
}

public void setState(String state){
this.state = state;
}

public String getState(){
return state;
}

@Override public boolean equals(Object other) { return EqualsBuilder.reflectionEquals(this, other); }

@Override
public String toString()
{
   return ToStringBuilder.reflectionToString(this); 
}
}

//Exclude fields
EqualsBuilder(this, other, new String[]{"id", "name"})

Annotation:

@JsonInclude(JsonInclude.Include.NON_EMPTY)
表示只有非空域才会进入json文件，如结果变为
{"customerId":555,"address":{"city":"Seattle"}}

之前提过不能用构造器，如果要用，就要加入@JsonCreator和@JsonProperty

@JsonInclude(JsonInclude.Include.NON_EMPTY)
public class Address {

@JsonCreator
public Address(@JsonProperty("city") String city, @JsonProperty("state") String state){
this.state = state;
this.city = city;
}

public void setCity(String city){
this.city = city;
}

@JsonProperty
public String getCity(){
return city;
}

public void setState(String state){
this.state = state;
}

@JsonProperty
public String getState(){
return state;
}

private String city;
private String state;
}

fasterxml + AutoValue

@AutoValue
@JsonDeserialize(builder = AutoValue_Account.Builder.class)
public abstract class Account {
@JsonProperty("name")
public abstract String getName();

@JsonProperty("accountNumber")
public abstract String getAccountNumber();

public static Builder builder(){
return new AutoValue_Account.Builder();
}

@AutoValue.Builder
public abstract static class Builder {
@JsonProperty("name")
   public abstract Builder setName(String value);

   @JsonProperty("accountNumber")
   public abstract Builder setAccountNumber(String value);
   public abstract Account build();
}
}

@JsonIgnore 不加入Json生成中
ref:
fasterxml例子

Spring简介(二) -- AOP

Spring对比与Guice是稍微重量级框架。

Spring注解

详细见Java依赖注入标准

Spring AOP

AOP (Aspect-oriented programming)是面向侧面的程序设计，是植入横切关注点，对程序做某一点的关注，如关注一个方法的运行时间。
AspectJ采用编译期织入和类加载期织入的方式织入切面，是语言级的AOP实现。AspectJ利用反射的方法拦截方法，对方法调用前后计算时间来实现。



ref:
AspectJ

Mac快捷键

Mac Keyboard	Win Keyboard(swap ctrl&command)
Control + Shift + Power	Control + Alt + Scroll Lock	锁屏
Command+space	Ctrl+Space	全局搜索Spot Search
Fn+Down	Page Down	下一页
Fn+Delete	Delete	删除游标后的字符
Ctrl+Space	win+Space	切换输入法
Fn+F11	F11	显示桌面
Command+tab	Ctrl+tab	app切换
Command+win	ctrl+win	tab切换
Command+win+F	ctrl+win+F	全屏/退出全屏
Command+M	ctrl+M	最小化窗口
	win+mouse scroll	放大word
Command+Plus	ctrl+Plus	放大Chrome
command+shift+4	ctrl+shift+4	截屏
command+shift+r/shift reload button		hard refresh无cache的加载
⌘command
⌥option
⇧shift
⇥tab
⌃control
命令行
Ctrl+R	win+R	搜索历史命令
Ctrl+A	win+A	游标回到头
Ctrl+b		上一页
Ctril+f		下一页
Ctril+d		删除改行
shift+g	shift+g	打开的文件跳到末尾
shift+z+z	shift+z+z	vim保存且退出(=:wq)
shift+z+q	shift+z+q	vim不保存且退出(=:q)
dd		vim删除一行(非插入状态)
p		vim复制刚才dd剪切的一行(非插入状态)
IntelliJ
command+O	Ctrl+O	查找文件
command+alt+右箭头	ctrl+alt+右箭头	后一步Navigate forward
command+alt+左箭头	ctrl+alt+左箭头	前一步Navigate backward
option+回车		加入import解决symbol

Linux常用命令

grep是查找某个关键字或正则表达式在某个文件某一行的匹配，结果为匹配的行及文件

返回含abc的匹配行：
zgrep "abc" application.log

在多个文件中查找：*.txt或者{}.txt
grep 'linux' {a,b}.txt

先返回含Exception的匹配行再返回含Hulu的匹配行，也就是同时含这两次的匹配行：
zgrep "Exception" */application*.log | grep "Hulu" 在application log中含Exception以及Hulu

按文件统计出现行数 -c
grep -c 'linux' *.txt

统计全部文件总行数=次数：
zgrep "Exception" */application*.log | grep "Hulu" | wc -l 数出现次数

关键词也可以为正则表达式：
vim中 /main 搜索main关键字，可以带正则表达式：如Time=[\d]{4,4}\.[\d]* ms
   按键盘n表示下一个搜索结果，按键盘N表示上一个搜索结果。
grep也可以带正则表达式：grep -c 'Time=[\d]{4,4}\.[\d]* ms' *service
grep 'linux\|win' a.txt

grep -c 'Time=[\d]{4,4}\.[\d]* ms' *service| sort | uniq -c | sort -rn |head 按文件统计词频且按大到小排序

grep是行匹配，而awk是词匹配（按空格）
打印所有行的第一个词（默认按空格划分），其他词不输出
awk '{print $1}' a.txt

自定义分隔符，如逗号
awk -F',' 'print $2' a.txt

排序
sort -r 递降排序结果
sort -u 去重且排序

进程操作
netstat -tupln | grep 1066	得到要终结进程的id
kill -9 16085	删除进程by id
ps aux | grep 1066	查看谁在用1066端口
查找操作
grep -color=auto "the" foo.txt	[文件内容]在某一个文件查找某字符串并打印结果行 -i 大小写敏感 -A 打印后几行 -B 前几行 -C 周围几行 -P 匹配正则表达式 "url=.*?;" 问号表示分号不加入匹配结果 -o 只显示匹配string -c 匹配结果行数 或|wc -l -v 不含某字符串 -n 显示行号 grep -r "the" * 递归在所有文件中查找某字符串 grep -A 3 -i "moon" foo.txt 在某一个文件查找一个字符串并打印紧接的3行   | head 查看前10行   | sed "s/my/your/" 替换（用your替my）
find . -iname "odin-get"	[文件名]搜索文件名通过名字 find ./Test -user kk 搜索文件名通过用户
文件操作
less foo.txt	高效率打开大文件。CTRL+F下一页，CTRL+B上一页. SHIFT+G文件底部
cat > foo.txt	创建文件。CTRL+D保存文件
cat foo.txt	查看文件
vim foo.txt	创建/编辑一个文件 /main 搜索main关键字 i/insert 插入模式，ESC退出插入模式 :q 或者Ctrl+z+q不保存退出 :wq 保存退出
rm foo.txt	删除文件
mv -i foo.txt Test/foo.txt	移动/重命名文件
cp -i foo.txt Test/foo.txt	复制一个文件
scp ~/a.txt jiajicen@vtasters.com:~/a.txt	复制一个文件到另一个host
scp -r jiajicen@vtasters.com:~/test ~/test	复制一个文件夹到另一个host. test文件夹不需要预先创建
tail -f app.log	即时打印最新的log，方便同步debug
目录操作
cd ..	/root/Docements 绝对路径 ./path 当前目录 ../path 父目录 ~/Desktop 根目录/Users/kk/Desktop ~john/bin 用户目录/Users/john/bin
ls -ltr	-ltr 按时间排序 -lh 显示大小权限 -alh 显示隐藏文件
pwd	显示当前目录
mkdir Test	创建目录
rm -r Test	删除目录
环境变量
export PATH="/home/Application/Java/bin:$PATH"	加入环境变量到内存，然后执行source加入到硬盘再重启命令行。 一次性操作为export PATH=$PATH:/home/Application/Java/bin >> ~/.bash_profile
source ~/.bash_profile (Mac)	加入到硬盘。Linux为~/.zshrc
echo $HOME	打印某环境变量
which java	得到某个命令如java路径，也是查找在~/.zshrc此命令的路径
系统操作
clear	清屏
sudo reboot	重启
sh abc.sh	执行一个程序，可能需要先提高权限chmod +x abc.sh 有时需要加点如./mvn才能执行，通过加入到.bash_profile可以避免
su Java	su=sudo, 用超级用户执行程序。
ssh vtasters.com	远程登录
nslookup	通过IP查找其hostname

agrep 文件每个block以EOF结束，agrep可以block查找


zshrc文件:
~/.zshrc
alias vm01="ssh ubuntu@172.31.29.246"

然后
echo "alias vps='ssh -X user@example.com'" >> ~/. zshrc
source ~/.zshrc
就可以输入vm01登录

Linux文件权限:
Linux权限为读写执行r, w, x.
Linux权限粒度为拥有者u，群组g，其他组o。
4,2,1表示读、写、执行权限。rwx=4+2+1=7, rw=4+2=6, rx=4+1=5

chmod 777 file.txt (<=> chmod u=rwx, g=rwx, o=rwx)
chmod 600 file.txt (<=> chmod u=rw, g=---, o=---)

查看字符

先copy到这个网站得到它的ascii码

然后查找

ssh问题

尝试ssh到某个host得到此错误Could not resolve host name但可以ping

尝试用ssh <host> -o ProxyCommand=none

Ref:
https://blog.csdn.net/u013197629/article/details/73608613

AWS目录

CloudTrail(AWS console user/API activities)

Athena

AWS KMS

Api gateway

产品	简介	类别	竞争者	用例
ElastiCache	Cache	Cache	Redis/Memcached
DynamoDB	NoSQL database	数据库	mongodb
Elasticsearch	多文档全文搜索	搜索	Lucene	Github, StackOverflow, HotelTonight, Wiki
S3	File store	文件存储	Disk/FTP
Lambda	事件驱动构件，无编程环境只与语言相关的EC2	serverless	Webtask	Alexa skill，微服务
Kinesis	实时发布者-订阅者系统	系统	Apache Kafka	股票实时统计 LinkedIn: 用户消息通知 游戏: 交易通知或预警
SQS	非实时处理任务系统，将同步变异步解耦削峰	系统	ActiveMQ	订单放入队列中
RDS	关系型数据库包括MySQL, MSSQL, Oracle等	数据库	各类数据库供应商
CloudWatch	server运维数据和警报	运维	Nagios
CloudTrail	AWS资源的变化和API数据	运维
CloudFormation	用json来描述AWS资源及其重要设置	运维
Pipeline	代码部署	运维
CloudFront	CDN service	系统
EMR	基于Hadoop来做数据计算的平台	系统
SES	电子邮件平台	邮件
AppSync	用于手机，网站的实时data sync服务	系统
Mobile Analytics	收集手机用量	系统
Athena	SQL分析服务	分析
Amplify	类似于Google Analytics分析，网站的点击数，时间等	系统

ElastiCache(=Cache)，DynamoDB(=数据库)，S3(=文件存储)。Cache最快当然也最贵，数据库可以建index，所以也比较快，价格比Cache稍便宜，而S3最慢也最便宜。

Alexa Skill例子

所有skill都继承Speechlet，onIntent处理每种intent，session是每次完整对话的内容。

public class ScoreKeeperSpeechlet implements Speechlet {
    @Override
    public SpeechletResponse onLaunch(final LaunchRequest request, final Session session)
            throws SpeechletException {

    }

    @Override
    public SpeechletResponse onIntent(IntentRequest request, Session session)
            throws SpeechletException {
        Intent intent = request.getIntent();
        if ("NewGameIntent".equals(intent.getName())) {

        } else if ("AddPlayerIntent".equals(intent.getName())) {

        }
    }

    @Override
    public void onSessionStarted(final SessionStartedRequest request, final Session session)
            throws SpeechletException {

    }

    @Override
    public void onSessionEnded(final SessionEndedRequest request, final Session session)
    }

}

ref:
https://github.com/amzn/alexa-skills-kit-java
这里链接有很多例子