npm install -D typescript
npx tsc --init 产生tsconfig.json
与网站结合一起用vite，vite包括安装typescript。my-app是文件夹名字
npm create vite@latest my-app -- --template vanilla-ts
cd my-app
npm install
npm run dev
每次改动跑这个命令：
cd my-app
npm run dev
http://localhost:5173/
The browser updates automatically
我遇到一个permission问题， 如果claude一直说找不到某个case，然后我让它写一个script返回总cases数量，然后我告诉他用这个script跑可以得到10000个，然后它自己debug出来了。所以需要教它debug
加入到.claude让每个feature或者bugfix来设定exit criteria
# Create a new venv with Python 3.11
C:\Users\KK\AppData\Local\Programs\Python\Python311\python.exe -m venv venv311
# Activate the venv
.\venv311\Scripts\activate # Windows PowerShell
pip install crowd-kit
python .\mle\multi_annotator_text_classification.py
f返回值是Awaitable, 就要用xxx_aync的函数
f返回值是Awaitable<bool>, 就要用if (await f)
self::functionA
return await EntAbc::queryFromReq($req)->queryMi->gen() |> vec ($$)
return await EntAbc::queryFromReq($req)->queryMi->gen() |> Vec\filter_nulls ($$)
Data Structure (Hack)
Vec: $a = vec[1, 2]
Dict: $b = dict["foot" => 2]
keyset: $c = keyset[1, 2]
const string USER_NAME = "we are";
static::USER_NAME
| Str\join | join成一个字符串 | $a = Str\join(Vec\reverse($vec), ','); |
| Str\format | 组成一个字符串 | $a = Str\format('%d is num', 2); |
| C\count | Vec的个数 | $a = C\count($requirements) |
| C\is_empty | Vec是否为空 | C\is_empty($requirement) |
| C\fb\any_async C\any | Vec中任意一个对应lambda返回true就return true | $has_deprecated_req = C\fb\any($requirements, async $requirement ==> {return true},); |
| Vec\map | Vec中每个对应lambda组成的新vec | $a = await Vec\map_async($requirements, async $requirement ==> {return await static::func(1)},); |
| Vec\filter | Vec中filter对应lambda组成的新vec | $a = Vec\filter($requirements, aysnc $requirement ==> 1 === 1,); |
| int->str | 类型转换 | $a = (string) $c |
| Vec\concat | 合并两个vec变成一个 | $c = Vec\concat($a, $b) |
| Vec\diff | 合并两个vec变成一个 | $c = Vec\diff($a, $b) |
| Vec\intersect | 求交集 | $c =Vec\intersect($a, $b) |
| C\count | Vec的个数 | $a = C\count($requirements) |
https://aws.amazon.com/solutions/implementations/video-on-demand-on-aws/
Audio on demean并没有具体教程，不过mediaConvert也可以覆盖。可以根据这个操作
https://docs.aws.amazon.com/mediaconvert/latest/ug/setting-up-audio-only.html
1. Input, remove the video selector
2. Add output groups, choose APPLE HLS
3. in the encoding setting in the output setting, remove video
4. in the Apple HLS, choose Container for Audio-only output, choose MPEG-2 Transport Stream (关键步骤）
AWS Data Pipeline用于ETL，一个use case是prod accout上的DDB数据clone到alpha account上DDB。先用Data Pipeline复制到alpha上的S3(CloudWatch 定时event)，然后再用lambda将数据加载到DDB(CloudWatch 定时event)。缺点是如果这两个步骤是独立的。
Glue有ApplyMapping等，如果需要对数据进行编辑，就应该用Glue
| "vtasters" | Exact Match, "use * card" |
| site:stackoverflow.com how to install java | 在某个网站查找 |
| Nasdaq news after:2020 | 时间范围查询 Nasdaq news before:2020-03-01 Nasdaq news on:2020-04-15 |
| fngu download filetype:pdf | 文件格式 |
| intitle:cloudwatch | 只搜title |
| allintitle:cloudwatch dashboard | 所有关键字都在title |
| related:amazon.com | 相似网站 |
Glue封装了PySpark和Spark SQL
DataSource0.count()
DataSource0.printSchema()
df = DataSource0.toDF()
找到value column中含数字字母的
df.filter(df['value'].rlike('\w+')).show()
找到value column中只含数字字母的
df.filter(df['value'].rlike('^a-zA-Z\d\s:') == False).show()
CDK是用多种语言实现的打包工具。
下面介绍是type script (类似于Node.js)
按这几个步骤初试
Step Function
Glue
可以用json来实现state，而不是用new tasks.GlueStartJobRun. 这是因为有些功能并不支持如GlueStartJobRun.sync.
const stateJson = {
Type: 'Task',
Resource: 'arn:aws:states:::dynamodb:putItem',
Parameters: {
TableName: table.tableName,
Item: {
id: {
S: 'MyEntry',
},
},
},
ResultPath: null,
};
Lambda
CDK中step function中的lambda支持payload参数
new tasks.LambdaInvoke(this, 'Invoke with payload field in the state input', {
lambdaFunction: fn,
payload: sfn.TaskInput.fromObject({
"execution.$": "$$.Excution.Id",
"catalogId": sfn.JsonPath.stringAt('$.catalogId'),
}),
});
API gateway
https://docs.aws.amazon.com/cdk/api/latest/docs/aws-apigateway-readme.html#integration-targets
产生StepFunction可以用console的template产生，节省时间
如何在Step function用变量
每一个task都有input和output，如果一个task的TaskStateEntered(AWS Console)的input是
{"input": {
"catalogId": "abc-cde"
}
}
State machine中，这样取值$.catalogId, 根目录对应的是input. 这个task也并不需要写ResultPath(只做filter之用).
input对应$.
output对应$.output.
content对应$..
通过选择，可以选取Lambda InputStream的输入，如
Parameters: {
"FunctionName": "xxx",
"Payload": {
"execution.$": "$$.Execution.Id",
"categoryId.$":: "$.catalogId",
"runId.$": "$output.JobRunState"
}
}
Payload对应Lambda的InputStream的输入
这个task(假设是lambda)是按照以下方式写入output:
public void handle(InputStream in, OutputStream out, Context context) {
HashMap map;
map.put("catalogId, "abc-cde");
String json = objectwriter.toJson(map);
out.write(json);
}
Map换成一个object也是一样的。
Context变量
Context含有step function execution arn，用的时候用$$.
"Execution.$": "$$.Execution.Id"
Step function能支持的Lambda的参数包括Payload，这是lambda的输入
"GetJobParams": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"Parameters": {
"FunctionName": "arn...Lambda",
"Payload": {
"execution.$": "$$.Execution.Id",
"catalogId.$": "$.catalogId"
}
}
}
"Glue StartJobRun": {
"Type": "Task",
"Resource": "arn:aws:states:::glue:startJobRun.sync",
"Parameters": {
"JobName": "my-etl-job"
},
"Next": "ValidateOutput"
},
Resource不能改，一般来说是用ARN，但Glue没有ARN，Step function通过JobName来定位Glue job。JobRunId不能用自定义格式或者不能加入作为参数，否则会说resource找不到。返回值是Id (JobRunId)，JobRunState，ErrorMessage, StartedOn等等。
如果某个Task failed还想继续执行下一个任务，可以用Catch，如ETL job失败，还是想将失败状态写入数据库
"Glue StartJobRun": {
"Type": "Task",
"Resource": "arn:aws:states:::glue:startJobRun.sync",
"Parameters": {
"JobName": "my-etl-job"
},
"Catch": [ {
"ErrorEquals": ["States.Timeout", "States.TaskFailed", "HandledError"],
"Next": "ValidateOutput"
} ],
"ResultPath": "$.output",
"Next": "ValidateOutput"
},
$.output存储Glue的输出结果，下一个task如lambda可以使用。
如何给一个task赋值
刚才讲到怎么调用变量，现在讲怎么给一个task的参数赋值。关键在于尾部加入.$
"Glue StartJobRun": {
"JobName": "my-etl-job"，
"Arguments": {
"--catalog_id.$": "$.catalogId"
}
},
Step function只支持这些参数
Lambda retry
lambda本身的retry并不支持，需要再step function里面定义retry
"Retry": [
{
"ErrorEquals": [
"States.ALL"
],
"IntervalSeconds": 30,
"MaxAttempts": 2,
"BackoffRate": 2
}
],
一步一步建立第一个API + lambda application
Lambda and API gateway Java example
API gateway也可以直接invoke Step function (默认为async的startExecution API，大概80ms latency，也可以invoke stopExcution, startSyncExecution），返回值为step function run id, start time等，一定要用Post。
Post在input里用$request.body支持request payload
IAM role，需要有APIGatewayInvokeFullAccess和StepFunctionFullAccess
Logging:
默认没有log，可以到Monitor->Logging去enable
需要自己创建一个log group
format需要加入$context.error.message, $context.integrationErrorMessage，否则即使是500，也不会知道是什么错误，其他log参数
Authentication & Authorization
可以用Federate + Cognito Identity pool/Cognito user pool 来Authenticate 用户，但这多数用于真正用户。另一种方法是用Lambda Authorizer作为interceptor来做验证。
作为程序用户，可以用IAM user（非AWS resource）或者IAM role（AWS resource）来验证。
需要AmazonAPIGatewayInvokeFullAccess的permission
Frontend用不checkin的方法来本地存API key
CORS
Request有两种：simple和non simple。我的case是返回contentType=json，不是Simple所以需要CORS support。API gateway设置如下：
allow-control-allow-orgin=https://www.vtasters.com (no slash)
access-control-allow-headers=access-control-allow-origin, content-type, authorization, referer
access-control-allow-methods=POST, GET
access-control-max-age=600 (Same as Chrome)
Document-based database. MongoDB wrapper.
大部分功能如index，连接方法类似于RDS，RDS上有full text search功能，支持词频搜索如shop coffee或coffee shop
Terms:
Collection: Table
Transaction:
multi-doc for 4.0 above
Other features
Change stream for a collection
Setup + Cloud 9 (IDE)
Shell命令
show dbs
show collections
use mydb (will create or switch to db)
db.createCollection("metadata") / db.mydb.createCollection("metadata")
db.metadata.insert({"name":"documentdb"})
db.metadata.findOne()
db.metadata.count()
db.metadata.find({name: "Netflix"})
db.metadata.find({name: /Netflix/}) -- like '%Netflix%' (case sensitive)
db.metadata.find({name: /^netflix$/i}) -- like '%netflix%' (case insensitive)
Java coding
import com.mongodb.MongoClient来获取数据
mongoClient.getDatabase('mydb');
DB connection
DB connection会看到是Database operation的两倍，因为会自动产生读和写两个connection，叫replica
https://docs.aws.amazon.com/documentdb/latest/developerguide/connect-to-replica-set.html
1000 per cluster: https://docs.aws.amazon.com/documentdb/latest/developerguide/limits.html
Glue是一个自动化的工具，有很多优点如自动生成script，支持常见ETL操作如ApplyMapping，BYOD custom script，crawler自动识别scheme等等。相比于DataPipeline和step functions属于更高层的封装。
Glue之初感
需要的IAM role:
S3FullAccess
AWSGlueServiceRole
CloudWatchLogsFullAccess
Transformation:
可以将CSV, TXT, TSV转化成JSON，JSON的形式不是List而是每个object并排如{"city": "b"}{"city": "a"}. 反之转换也可以，但JSON的输入形式也不能是list，否则ApplyMapping等不能识别。可以支持nested JSON(三层以上均可)
Crawler:
Crawler 自动可以crawl指定bucket里面的file的metadata如column names， file type，ski header，count等等。
Custom Scripts (Spark DataFrame)
Custom transformation可以插入自定义的script，直接integrate到ETL job，但需要实现指定的API。下面是一段例子：
df = dfc.select(list(dfc.keys())[0]).toDF()
df_filtered = df.filter(df["year"] > 2018)
dyf_filtered = DynamicFrame.fromDF(df_filtered, glueContext, "filter_votes")
return (DynamicFrameCollection({"CustomTransform0": dyf_filtered}, glueContext))
去除空的row：
df_filtered = df.filter("videoName != ''")
如果出现空行，强制输出错误，返回到Glue errorMessage
if df.count != df_filtered.count:
raise ValueError('Empty row detected.')
custom script 需要和SelectFromCollection连用
Trigger:
Lambda可以作为trigger
https://aws.amazon.com/premiumsupport/knowledge-center/start-glue-job-crawler-completes-lambda/
可以用一个job succeeded event来trigger另一个job，这样就形成一个workflow
Programming:
可视化和代码可以自由转换。首先，可视化创建ETL job，然后转到scripts就看到自动生成的代码。所以只要用这个代码，就可以反过来创建ETL job
要加log的话：logger = glueContext.get_logger()
Python: https://github.com/aws-samples/aws-glue-samples/blob/master/examples/data_cleaning_and_lambda.md
Integration with AWS service: 用boto3 library
Glue API:
支持Scala和Python
Python: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/glue.html
start_job_run
只要用下面命令glue start-job-run指明如job-name和scriptLocation等job参数就可以创建。返回JobRunId
$ aws glue start-job-run --job-name "CSV to CSV" --arguments='--scriptLocation="s3://my_glue/libraries/test_lib.py"'
还支持custom parameters比如--source_file_s3_path, --targetFileS3Path等最多达50个。这样就可以令pipeline更灵活和generic，支持不同的输入和输出以及变换操作。
用的时候是
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'source_file_s3_path']
logger.info('path name:' + args['source_file_s3_path'])
例子：https://stackoverflow.com/questions/52316668/aws-glue-job-input-parameters
还可以设置concurrency，这样可以同时跑不同parameters对应的job。
Job parameters for CDK:
https://awscdk.io/packages/@aws-cdk/aws-glue@1.22.0/#/./@aws-cdk_aws-glue.CfnJob
Default parameters:
https://docs.amazonaws.cn/en_us/glue/latest/dg/aws-glue-programming-etl-glue-arguments.html
get_job_runs
输入参数为jobName和jobRunId，返回JobRunState(Failed, Succeeded)和errorMessage
Shared code in Glue jobs by Python lib
https://medium.com/@bv_subhash/sharing-re-usable-code-across-multiple-aws-glue-jobs-290e7e8b3025
Glue error Code:
Glue Concurrency
Timeout
Custom error
Internal Failure
Deployment:
由于ETL的Python script都是保存在S3的，所以如果代码commit到git的话就要手动上传到S3。解决方案是利用CDK里面的Assets将本地代码上传到S3. Ref
DocumentDB:
Glue可以连documentDB，用于ETL，这里我们用来做update 一个record. 需要在glue手动设置DocDB的连接。
secrets_manager_client = boto3.client("secretsmanager", region_name="us-west-2")
workflow_status = [{
"_id": job_run_id,
"statux": "xxx"
}]
workflow_status_frame = DynamicFrame.fromDF(spark.createDataFrame(workflow_status), glueContext, "nested")
db_writer(workflow_status_frame, "my_db", "workflow_status")
db_writer(df, database_name, colection_name):
write_documentdb_options = {
"uri":
"database": database_name
"collection": colection_name
....
}
glueContext.write_dynamic_frame.from_options(df, connection_type="documentdb", connection_options=write_documentdb_options)
与Tableau和SSIS类似，用于生成BI report。
report的参数
先产生一个paramter和对应的filter，然后就可以在report的URL加入如下，jobId是parameter name
https://us-west-2.quicksight.aws.amazon.com/sn/dashboards/e0176d07-f509#p.jobId=1568446
用户互动输入filter
原理同上，不过再加一个control在parameter上，用于用户输入。Dataset在一个Athena的view上，view处理了一些数据。
[1] https://docs.aws.amazon.com/quicksight/latest/user/parameters-in-a-url.html
[2] https://aws.amazon.com/blogs/big-data/using-quicksight-parameters-and-controls-to-drive-interactivity-in-your-dashboards/
VisualVM - 分析机器上app，CPU，memory运行情况
Apache JMeter - load test 工具，用于人工生成traffic，重现问题
Youkit/Eclipse Profiler - 分析内存泄漏memory leak
本问题是一个网站有memory leak的问题，导致heap memory usage在deployment后都一直上升
步骤
1. 用JMeter产生traffic，且用VisualVM观察问题是否重现且确认收集够数据，就停止收集
2. copy heap dump到安装了Youkit的机器
3. 用Youkit分析memory。发现有大量duplicate objects(Youkit定义是equal field by field或者数组元素一样)。打开QueryLogReporter可以发现这些重复对象都是来源于某一个class。
分析代码后发现，我们用了HTTP intercepter来拦截所有request然后将这些metrics放入一个MetricsManager的对象中，而这些对象并没有有效被删除，导致在ThreadLocal上累积。
找到根本原因后，我们在代码中在每次收集HTTP request后调用这个manager的一个delete函数。
创建AWS账号时会自带一个default VPC。
1. 创建RDS时，用上这个VPC
2. 一开始一直Workbench连不上，后来发觉缺少几个设置。设点击主DB instance (非cluster)为publicly accessible。这个设置至关重要，因为根据描述，DB在VPC中，外部设备如workbench和其他VPC都不能访问。否则就要通过[1]的方法，通过SSH到同一VPC下EC2来进行访问。设置了public并不代表任何人都可以访问，因为VPC会有访问限制，详看下一步。
3. DB对应的security group（防火墙）inbound规则允许MySQL和3306端口并将我的机器IP (My IP选项或者custom选项用https://checkip.amazonaws.com/ 来查询)加入到白名单中。outbound不用修改。
4. 设置workbench
Hostname: 用主instance (writer)的hostname， **-us-west-2.rds.amazonaws.com
port: 3306
username: admin
pw: 创建实例时候记下的
Ref:
[1] https://www.inoneo.com/en/blog/15/amazon-aws/connect-to-an-aws-rds-instance-inside-a-vpc-using-mysql-workbench
1. Setup AWS Toolkit in Eclipse [1]
2. Setup AWS credentials [2]
The instructions are not quite clear. A few steps required here:
(1) Create an IAM user without any permissions boundaries
(2) Add permissions for Lambda, S3, CloudFormation and iam (Inline policy). For iam, the iam:CreateRole is required [3]. But I added full access.
Note: if there is any error in deploying the lambda, check the missing permssions in CloudFormation events or the AWS console in Eclipse.
3. Create a AWS serverless Java project with hello-world blueprint [4]
4. Deploy the project
5. Test lambda. In lambda service in AWS console (Be sure in the right region), add test event
"TestInput"
The run the lambda. Check the logs in CloudWatch logs.
Ref:
[1] https://docs.aws.amazon.com/toolkit-for-eclipse/v1/user-guide/setup-install.html
[2] https://docs.aws.amazon.com/toolkit-for-eclipse/v1/user-guide/setup-credentials.html
[4] https://docs.aws.amazon.com/toolkit-for-eclipse/v1/user-guide/serverless-projects.html
Write-through（直写模式）在数据更新时，同时写入缓存Cache和后端存储。此模式的优点是操作简单；缺点是因为数据修改需要同时写入存储，数据写入速度较慢。
Write-back（回写模式）在数据更新时只写入缓存Cache。只在数据被替换出缓存时，被修改的缓存数据才会被写到后端存储。此模式的优点是数据写入速度快，因为不需要写存储；缺点是一旦更新后的数据未被写入存储时出现系统掉电的情况，数据将无法找回。
