冷月_wjh
TA的个人档案
冷月_wjh_个人页
文章
12
问答
0
视频
0
个人介绍
暂无个人介绍
擅长的技术
- Java
- Linux
- 数据库
-
发表了文章 2020-12-21
一站式极致读吞吐，加速业务数字化转型 - 阿里云对象存储 OSS加速器解析与实践
OSS加速器介绍 对象存储OSS（Object Storage Service）具有海量、可靠、安全、高性能、低成本的特点。OSS提供标准、低频、归档、冷归档类型，覆盖多种数据从热到冷的存储需求，单个文件的大小从1字节到48.8TB，可以存储的文件个数无限制。OSS已成为互联网、企业级数据应用的基础设施。 然而，随着互联网业务的发展，越来越多的业务对数据的吞吐提出了更高的要求。为此，阿里云对象存储OSS重磅推出OSS加速器功能，它可以缓存OSS中的热点对象，提供高性能、高吞吐量的数据访问服务。 OSS加速器适用于基因训练、机器学习、数据湖大数据计算等需要大带宽，且数据重复读较多的场景。OSS加速器是一个标准的服务端缓存服务，和计算完全解耦，不同于传统的OSS功能，其为AZ级的服务模式，用户可以在相关AZ创建加速器，计算引擎通过缓存加速域名进行数据读写。在典型的一些数据湖场景中，例如大数据计算(EMR/DLA等) + OSS场景中，数据读取要求的带宽可能高达数百Gbps~Tbps，普通存储空间的吞吐能力往往无法轻松应对这种大带宽的读取需求。这时您可以开启OSS加速器，将需要重复读取的数据缓存在加速中。当上层应用向OSS加速器请求数据时，加速器根据您购买的加速器的容量大小提供1.6Gbps/TB的带宽（或200MBps/TB的吞吐）和一定的回源带宽(回源带宽是指首次访问不存在于加速器中的数据时，从OSS读取原始数据的带宽，然后OSS会将该数据缓存到加速器中)。 下图是数据访问流程： OSS加速器技术特点 1) 极致吞吐能力 OSS加速器有效解决多种应用场景（比如数据湖等）的读吞吐的挑战，特别是重复读的场景，其能够每TB提供200MBps（Byte）的吞吐能力，线性扩展，最高可以交付Tbps的带宽交付能力。2）弹性伸缩 通常计算任务有周期性需求，如何有效避免资源浪费，提效降本？通过OSS加速器的弹性伸缩能力，能够在线进行扩容缩容，可以按需释放或提升资源，有效降低TCO。3）存算分离 OSS加速器满足计算资源和存储资源解耦，面对计算任务的多样化，不再需要多个自建缓存搭建匹配，存算分离，灵活选择引擎和版本，满足多业务场景的吞吐加速。4）数据一致 同时，基于OSS智能元数据架构，OSS加速器提供了传统缓存方案不具备的一致性，当OSS上文件被更新时，加速器能自动识别，确保引擎读取到的都是最新数据。 使用示例 下面将演示如何创建OSS加速器以及使用OSS加速器的实际效果。 创建OSS加速器 用户可以在OSS控制台主页面左侧的导航栏下方看到"OSS加速器"标签页，申请试用通过后，可以看到OSS加速器功能界面 点击“创建OSS加速器”按钮后我们可以创建OSS加速器，这个时候会弹出"创建OSS加速器"对话框 您可以设置加速器的名称和容量，目前加速器的容量是20TB起步，最高可以设置100TB(更高的容量需要开工单进行申请)，容量步长为5TB。目前OSS加速器支持的的可用区是上海的cn-shanghai-g可用区，其他的可用区正在逐步开放中。 修改加速器容量 创建好加速器后，我们还可以修改加速器的容量，修改加速器的容量同样会修改相关的带宽(吞吐)指标（这里显示的是MB/s，可以自行折算成Mbps或Gbps）。 设置加速路径 创建好加速器后，我们需要设置加速路径(加速路径是指当访问对象的前缀匹配加速路径时，会访问加速器进行加速数据访问)。 目前一个加速器可以可以配置的bucket数量无限制，每个bucket可以配置最多10个加速路径(如果不设置加速路径的话，将会加速整个bucket里面的数据读取)。 效果展示 下面我们展示一下加速容量从40TB提升到100TB，然后再降为70TB的效果图(图上的带宽吞吐单位是Byte/s，读取的对象之前访问过，已经缓存在加速器中)。 40TB的加速器容量对应的读带宽约是62.5Gbps，当我们将加速器容量提升到100TB时，带宽上升到约156.25Gbps，然后我们将加速器容量降低到70TB，我们可以看到带宽降到约109.375Gbps。可见OSS加速器可以很好地实现读吞吐带宽的按需提升，弹性伸缩，满足相关业务的需求。 试用申请 目前阿里云对象存储OSS加速器在邀测使用中，阿里云企业认证用户可申请试用，点击立即申请
-
发表了文章 2019-07-15
如何将Elasticsearch的快照备份至OSS
前言 Elasticsearch 是一个开源的分布式 RESTful 搜索和分析引擎。它可以在近实时条件下，存储，查询和分析海量的数据。它还支持将快照备份至HDFS/S3上面，而阿里云OSS兼容S3的API，本文将介绍如何使用ES的Repository-S3插件将快照备份至OSS。 部署与配置 首先，我们需要安装repository-s3，可以参考官方文档：https://www.elastic.co/guide/en/elasticsearch/plugins/7.2/repository-s3.html 启动ES，我们可以从log中看到，ES已经load了这个plugin： [2019-07-15T14:12:09,225][INFO ][o.e.p.PluginsService ] [master] loaded module [aggs-matrix-stats] [2019-07-15T14:12:09,225][INFO ][o.e.p.PluginsService ] [master] loaded module [analysis-common] [2019-07-15T14:12:09,225][INFO ][o.e.p.PluginsService ] [master] loaded module [ingest-common] [2019-07-15T14:12:09,226][INFO ][o.e.p.PluginsService ] [master] loaded module [ingest-geoip] [2019-07-15T14:12:09,226][INFO ][o.e.p.PluginsService ] [master] loaded module [ingest-user-agent] [2019-07-15T14:12:09,226][INFO ][o.e.p.PluginsService ] [master] loaded module [lang-expression] [2019-07-15T14:12:09,226][INFO ][o.e.p.PluginsService ] [master] loaded module [lang-mustache] [2019-07-15T14:12:09,227][INFO ][o.e.p.PluginsService ] [master] loaded module [lang-painless] [2019-07-15T14:12:09,227][INFO ][o.e.p.PluginsService ] [master] loaded module [mapper-extras] [2019-07-15T14:12:09,227][INFO ][o.e.p.PluginsService ] [master] loaded module [parent-join] [2019-07-15T14:12:09,227][INFO ][o.e.p.PluginsService ] [master] loaded module [percolator] [2019-07-15T14:12:09,227][INFO ][o.e.p.PluginsService ] [master] loaded module [rank-eval] [2019-07-15T14:12:09,228][INFO ][o.e.p.PluginsService ] [master] loaded module [reindex] [2019-07-15T14:12:09,228][INFO ][o.e.p.PluginsService ] [master] loaded module [repository-url] [2019-07-15T14:12:09,228][INFO ][o.e.p.PluginsService ] [master] loaded module [transport-netty4] [2019-07-15T14:12:09,228][INFO ][o.e.p.PluginsService ] [master] loaded plugin [repository-s3] [2019-07-15T14:12:12,375][INFO ][o.e.d.DiscoveryModule ] [master] using discovery type [zen] and seed hosts providers [settings] [2019-07-15T14:12:12,801][INFO ][o.e.n.Node ] [master] initialized [2019-07-15T14:12:12,802][INFO ][o.e.n.Node ] [master] starting ... 然后，我们需要将OSS使用的Access Key和Secret Key配置到ES去，分别执行下面的命令： bin/elasticsearch-keystore add s3.client.default.access_key bin/elasticsearch-keystore add s3.client.default.secret_key 运行 首先，我们创建一个备份： [root@master ~]# curl -XPUT 'http://localhost:9200/_snapshot/test' -H 'Content-Type: application/json' -d '{ "type": "s3", "settings": { "bucket": "hadoop-oss-test", "endpoint": "oss-cn-zhangjiakou-internal.aliyuncs.com"} }' {"acknowledged":true} NOTE: 上面的命令默认使用https协议来传输数据，如果想使用http协议，需要将"protocol": "http", "disable_chunked_encoding": true加到settings里面(这个特性将会在新版本发布后可用)。 可以使用下面的命令来确实创建是否成功： [root@master ~]# curl -XGET localhost:9200/_snapshot/test?pretty { "test" : { "type" : "s3", "settings" : { "bucket" : "hadoop-oss-test", "endpoint" : "oss-cn-zhangjiakou-internal.aliyuncs.com" } } } 我们可以写入一些测试数据到ES，然后看下目前集群的索引信息: [root@master ~]# curl -X GET "localhost:9200/_cat/indices?v" health status index uuid pri rep docs.count docs.deleted store.size pri.store.size green open sales 89ouBy6RQsuT34QRbn_jeQ 10 0 271786 0 15mb 15mb green open customer fQCMEvXsQOu0UgMm1SAJlA 5 0 10000 0 717kb 717kb 假设我们只备份sales索引： [root@master ~]# curl -XPUT 'http://localhost:9200/_snapshot/test/sales' -H 'Content-Type: application/json' -d '{ "indices": "sales" }' {"accepted":true} 然后我们可以从OSS控制台看到备份的结果： 现在我们再往sales索引里面写一些数据： [root@master ~]# curl -X GET "localhost:9200/_cat/indices?v" health status index uuid pri rep docs.count docs.deleted store.size pri.store.size green open sales 89ouBy6RQsuT34QRbn_jeQ 10 0 281502 0 15.6mb 15.6mb green open customer fQCMEvXsQOu0UgMm1SAJlA 5 0 10000 0 717kb 717kb 我们利用刚才备份到OSS的快照来恢复sales索引，分别执行下面的命令： [root@master ~]# curl -XPOST localhost:9200/sales/_close {"acknowledged":true,"shards_acknowledged":true,"indices":{"sales":{"closed":true}}} [root@master ~]# curl -XPOST 'http://localhost:9200/_snapshot/test/sales/_restore?pretty' { "accepted" : true } [root@master ~]# curl -X GET "localhost:9200/_cat/indices?v" health status index uuid pri rep docs.count docs.deleted store.size pri.store.size green open sales 89ouBy6RQsuT34QRbn_jeQ 10 0 271786 0 15mb 15mb green open customer fQCMEvXsQOu0UgMm1SAJlA 5 0 10000 0 717kb 717kb 我们可以看到，sales索引跟之前的一致。 参考资料 https://www.elastic.co/guide/en/elasticsearch/plugins/7.2/repository-s3.html https://www.elastic.co/cn/products/elasticsearch
-
发表了文章 2019-04-23
使用Fluentd读写OSS
前言 Fluentd是一个实时开源的数据收集器，基于CRuby实现，td-agent是其商业化版本，由Treasure Data公司维护。本文将介绍如何使Fluentd能够读写OSS。 安装 首先下载并安装td-agent，笔者使用的是td-agent-3.3.0-1.el7.x86_64.rpm，使用rpm命令安装: [root@apache ~]# rpm -ivh td-agent-3.3.0-1.el7.x86_64.rpm 然后，需要安装Fluentd的OSS plugin: [root@apache ~]# /usr/sbin/td-agent-gem install fluent-plugin-aliyun-oss 这里请注意，因为我们使用的是td-agent，安装Fluentd plugin时需要使用td-agent的td-agent-gem(/usr/sbin/td-agent-gem)。原因是td-agent有自己的Ruby，你需要将plugin安装到它的Ruby里面，而不是其他的Ruby，否则将会找不到已经安装好的plugin。 具体可以参见Fluentd的官方文档 安装完成后，我们可以查看安装的OSS plugin: [root@apache ~]# /usr/sbin/td-agent-gem list fluent-plugin-aliyun-oss *** LOCAL GEMS *** fluent-plugin-aliyun-oss (0.0.1) fluent-plugin-aliyun-oss这个plugin包含两部分： Fluent OSS output plugin将数据缓存在本地，达到设定的条件后，将缓存的数据(压缩后，如果设置的话)上传到OSS。 Fluent OSS input plugin首先，OSS的bucket需要配置事件通知，这篇文章介绍了如何设置，得到MNS的Queue与Endpoint。设置好之后，这个plugin会定时地从MNS拉取消息，从消息中获取上传的Objects，最后从OSS读取这些Objects，再发往下游。 下面将分别介绍如何配置，具体的配置参数说明参见github: https://github.com/aliyun/fluent-plugin-oss 配置(向OSS写数据) 下面是一个例子，将读到的数据，每分钟一个文件写到OSS中。其中endpoint/bucket/access_key_id/access_key_secret是必填项，其他的都是可选项。 <system> workers 6 </system> <match input.*> @type oss endpoint <OSS endpoint to connect to> bucket <Your Bucket> access_key_id <Your Access Key> access_key_secret <Your Secret Key> upload_crc_enable false path "fluent-oss/logs" auto_create_bucket true key_format "%{path}/%{time_slice}/events_%{index}_%{thread_id}.%{file_extension}" #key_format %{path}/events/ts=%{time_slice}/events_%{index}_%{thread_id}.%{file_extension} time_slice_format %Y%m%d-%H store_as gzip <buffer tag,time> @type file path /var/log/fluent/oss/${ENV['SERVERENGINE_WORKER_ID']} timekey 60 # 1 min partition timekey_wait 1s # timekey_use_utc true flush_thread_count 1 </buffer> <format> @type json </format> </match> 我们可以从OSS控制台看到效果: 配置(从OSS读数据) 下面是一个配置示例，其中endpoint/bucket/access_key_id/access_key_secret和MNS的endpoint/queue是必填项，其他的都是可选项。 <source> @type oss endpoint <OSS endpoint to connect to> bucket <Your Bucket> access_key_id <Your Access Key> access_key_secret <Your Secret Key> store_local false store_as gzip flush_batch_lines 800 flush_pause_milliseconds 1 download_crc_enable false <mns> endpoint <MNS endpoint to connect to, E.g.,{account-id}.mns.cn-zhangjiakou-internal.aliyuncs.com> queue <MNS queue> poll_interval_seconds 1 </mns> <parse> @type json </parse> </source> 我们可以从log中看一下运行状态 2019-04-23 15:38:14 +0800 [info]: #5 start to poll message from MNS queue fluentd-oss 2019-04-23 15:38:14 +0800 [info]: #5 http://1305310278558820.mns.cn-zhangjiakou-internal.aliyuncs.com/queues/fluentd-oss/messages 2019-04-23 15:38:14 +0800 [info]: #5 read object fluent-oss/logs/20190423-12/events_10_70226640160100.gz, size 4389548 from OSS 2019-04-23 15:38:15 +0800 [info]: #1 http://1305310278558820.mns.cn-zhangjiakou-internal.aliyuncs.com/queues/fluentd-oss/messages?ReceiptHandle=0BC1EA4E51483D4EAC69736941044AAE-MjY5ODkgMTU1NjAwNTAwODMwNSAzNjAwMDA 2019-04-23 15:38:16 +0800 [info]: #1 start to poll message from MNS queue fluentd-oss 2019-04-23 15:38:16 +0800 [info]: #1 http://1305310278558820.mns.cn-zhangjiakou-internal.aliyuncs.com/queues/fluentd-oss/messages 2019-04-23 15:38:16 +0800 [info]: #1 read object fluent-oss/logs/20190423-09/events_50_69939581261780.gz, size 6750045 from OSS 参考资料 https://github.com/aliyun/fluent-plugin-osshttps://rubygems.org/gems/fluent-plugin-aliyun-osshttps://www.fluentd.org/https://docs.fluentd.org/v1.0/articles/quickstart
-
发表了文章 2019-03-01
Apache Hadoop 2.7如何支持读写OSS
背景 2017.12.13日Apache Hadoop 3.0.0正式版本发布，默认支持阿里云OSS对象存储系统，作为Hadoop兼容的文件系统，后续版本号大于等于Hadoop 2.9.x系列也支持OSS。然而，低版本的Apache Hadoop官方不再支持OSS，本文将描述如何通过支持包来使Hadoop 2.7.2能够读写OSS。 如何使用 下面的步骤需要在所有的Hadoop节点执行 下载支持包 http://gosspublic.alicdn.com/hadoop-spark/hadoop-oss-2.7.2.tar.gz 解压这个支持包，里面的文件是： [root@apache hadoop-oss-2.7.2]# ls -lh 总用量 3.1M -rw-r--r-- 1 root root 3.1M 2月 28 17:01 hadoop-aliyun-2.7.2.jar 这个支持包是根据Hadoop 2.7.2的版本，并打了Apache Hadoop对OSS支持的patch后编译得到，其他的小版本对OSS的支持后续也将陆续提供。 部署 首先将文件hadoop-aliyun-2.7.2.jar复制到$HADOOP_HOME/share/hadoop/tools/lib/目录下； 修改$HADOOP_HOME/libexec/hadoop-config.sh文件，在文件的327行加下代码： CLASSPATH=$CLASSPATH:$TOOL_PATH 修改的目的就是将$HADOOP_HOME/share/hadoop/tools/lib/放到Hadoop的CLASSPATH里面；下面是修改前后，这个文件的diff供参考(hadoop-config.sh.bak是修改前的文件): [root@apache hadoop-2.7.2]# diff -C 3 libexec/hadoop-config.sh.bak libexec/hadoop-config.sh *** libexec/hadoop-config.sh.bak 2019-03-01 10:35:59.629136885 +0800 --- libexec/hadoop-config.sh 2019-02-28 16:33:39.661707800 +0800 *************** *** 325,330 **** --- 325,332 ---- CLASSPATH=${CLASSPATH}:$HADOOP_MAPRED_HOME/$MAPRED_DIR'/*' fi + CLASSPATH=$CLASSPATH:$TOOL_PATH + # Add the user-specified CLASSPATH via HADOOP_CLASSPATH # Add it first or last depending on if user has # set env-var HADOOP_USER_CLASSPATH_FIRST 增加OSS的配置 修改core-site.xml文件，增加如下配置项： 配置项 值 说明 fs.oss.endpoint 如 oss-cn-zhangjiakou-internal.aliyuncs.com 要连接的endpoint fs.oss.accessKeyId access key id fs.oss.accessKeySecret access key secret fs.oss.impl org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem hadoop oss文件系统实现类，目前固定为这个 fs.oss.buffer.dir /tmp/oss 临时文件目录 fs.oss.connection.secure.enabled false 是否enable https, 根据需要来设置，enable https会影响性能 fs.oss.connection.maximum 2048 与oss的连接数，根据需要设置 相关参数的解释可以在这里找到 重启集群，验证读写OSS 增加配置后，重启集群，重启后，可以测试 # 测试写 hadoop fs -mkdir oss://{your-bucket-name}/hadoop-test # 测试读 hadoop fs -ls oss://{your-bucket-name}/ 运行teragen [root@apache hadoop-2.7.2]# hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.2.jar teragen -Dmapred.map.tasks=100 10995116 oss://{your-bucket-name}/1G-input 19/02/28 16:38:59 INFO client.RMProxy: Connecting to ResourceManager at apache/192.168.0.176:8032 19/02/28 16:39:01 INFO terasort.TeraSort: Generating 10995116 using 100 19/02/28 16:39:01 INFO mapreduce.JobSubmitter: number of splits:100 19/02/28 16:39:01 INFO Configuration.deprecation: mapred.map.tasks is deprecated. Instead, use mapreduce.job.maps 19/02/28 16:39:01 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1551343125387_0001 19/02/28 16:39:02 INFO impl.YarnClientImpl: Submitted application application_1551343125387_0001 19/02/28 16:39:02 INFO mapreduce.Job: The url to track the job: http://apache:8088/proxy/application_1551343125387_0001/ 19/02/28 16:39:02 INFO mapreduce.Job: Running job: job_1551343125387_0001 19/02/28 16:39:09 INFO mapreduce.Job: Job job_1551343125387_0001 running in uber mode : false 19/02/28 16:39:09 INFO mapreduce.Job: map 0% reduce 0% 19/02/28 16:39:18 INFO mapreduce.Job: map 1% reduce 0% 19/02/28 16:39:19 INFO mapreduce.Job: map 2% reduce 0% 19/02/28 16:39:21 INFO mapreduce.Job: map 4% reduce 0% 19/02/28 16:39:25 INFO mapreduce.Job: map 5% reduce 0% 19/02/28 16:39:28 INFO mapreduce.Job: map 6% reduce 0% ...... 19/02/28 16:42:36 INFO mapreduce.Job: map 94% reduce 0% 19/02/28 16:42:38 INFO mapreduce.Job: map 95% reduce 0% 19/02/28 16:42:41 INFO mapreduce.Job: map 96% reduce 0% 19/02/28 16:42:44 INFO mapreduce.Job: map 97% reduce 0% 19/02/28 16:42:45 INFO mapreduce.Job: map 98% reduce 0% 19/02/28 16:42:46 INFO mapreduce.Job: map 99% reduce 0% 19/02/28 16:42:48 INFO mapreduce.Job: map 100% reduce 0% 19/02/28 16:43:11 INFO mapreduce.Job: Job job_1551343125387_0001 completed successfully 19/02/28 16:43:12 INFO mapreduce.Job: Counters: 37 File System Counters FILE: Number of bytes read=0 FILE: Number of bytes written=11931190 FILE: Number of read operations=0 FILE: Number of large read operations=0 FILE: Number of write operations=0 HDFS: Number of bytes read=8497 HDFS: Number of bytes written=0 HDFS: Number of read operations=100 HDFS: Number of large read operations=0 HDFS: Number of write operations=0 OSS: Number of bytes read=0 OSS: Number of bytes written=1099511600 OSS: Number of read operations=1100 OSS: Number of large read operations=0 OSS: Number of write operations=500 ...... 运行distcp 从OSS往HDFS拷贝数据 [root@apache hadoop-2.7.2]# hadoop distcp oss://{your-bucket-name}/data hdfs:/data/input 19/03/05 09:43:59 INFO tools.DistCp: Input Options: DistCpOptions{atomicCommit=false, syncFolder=false, deleteMissing=false, ignoreFailures=false, maxMaps=20, sslConfigurationFile='null', copyStrategy='uniformsize', sourceFileListing=null, sourcePaths=[oss://{your-bucket-name}/data], targetPath=hdfs:/data/input, targetPathExists=false, preserveRawXattrs=false} 19/03/05 09:43:59 INFO client.RMProxy: Connecting to ResourceManager at apache/192.168.0.176:8032 19/03/05 09:44:00 INFO Configuration.deprecation: io.sort.mb is deprecated. Instead, use mapreduce.task.io.sort.mb 19/03/05 09:44:00 INFO Configuration.deprecation: io.sort.factor is deprecated. Instead, use mapreduce.task.io.sort.factor 19/03/05 09:44:01 INFO client.RMProxy: Connecting to ResourceManager at apache/192.168.0.176:8032 19/03/05 09:44:01 INFO mapreduce.JobSubmitter: number of splits:24 19/03/05 09:44:01 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1551343125387_0008 19/03/05 09:44:01 INFO impl.YarnClientImpl: Submitted application application_1551343125387_0008 19/03/05 09:44:01 INFO mapreduce.Job: The url to track the job: http://apache:8088/proxy/application_1551343125387_0008/ 19/03/05 09:44:01 INFO tools.DistCp: DistCp job-id: job_1551343125387_0008 19/03/05 09:44:01 INFO mapreduce.Job: Running job: job_1551343125387_0008 19/03/05 09:44:07 INFO mapreduce.Job: Job job_1551343125387_0008 running in uber mode : false 19/03/05 09:44:07 INFO mapreduce.Job: map 0% reduce 0% 19/03/05 09:44:16 INFO mapreduce.Job: map 4% reduce 0% 19/03/05 09:44:19 INFO mapreduce.Job: map 8% reduce 0% ...... 19/03/05 09:45:11 INFO mapreduce.Job: map 96% reduce 0% 19/03/05 09:45:12 INFO mapreduce.Job: map 100% reduce 0% 19/03/05 09:45:13 INFO mapreduce.Job: Job job_1551343125387_0008 completed successfully 19/03/05 09:45:13 INFO mapreduce.Job: Counters: 38 File System Counters FILE: Number of bytes read=0 FILE: Number of bytes written=2932262 FILE: Number of read operations=0 FILE: Number of large read operations=0 FILE: Number of write operations=0 HDFS: Number of bytes read=24152 HDFS: Number of bytes written=1099511600 HDFS: Number of read operations=898 HDFS: Number of large read operations=0 HDFS: Number of write operations=251 OSS: Number of bytes read=1099511600 OSS: Number of bytes written=0 OSS: Number of read operations=2404 OSS: Number of large read operations=0 OSS: Number of write operations=0 ...... [root@apache hadoop-2.7.2]# hadoop fs -ls hdfs:/data Found 1 items drwxr-xr-x - root supergroup 0 2019-03-05 09:45 hdfs:///data/input 从HDFS往OSS拷贝数据 [root@apache hadoop-2.7.2]# hadoop distcp hdfs:/data/input oss://{your-bucket-name}/data/output 19/03/05 09:48:06 INFO tools.DistCp: Input Options: DistCpOptions{atomicCommit=false, syncFolder=false, deleteMissing=false, ignoreFailures=false, maxMaps=20, sslConfigurationFile='null', copyStrategy='uniformsize', sourceFileListing=null, sourcePaths=[hdfs:/data/input], targetPath=oss://{your-bucket-name}/data/output, targetPathExists=false, preserveRawXattrs=false} 19/03/05 09:48:06 INFO client.RMProxy: Connecting to ResourceManager at apache/192.168.0.176:8032 19/03/05 09:48:06 INFO Configuration.deprecation: io.sort.mb is deprecated. Instead, use mapreduce.task.io.sort.mb 19/03/05 09:48:06 INFO Configuration.deprecation: io.sort.factor is deprecated. Instead, use mapreduce.task.io.sort.factor 19/03/05 09:48:07 INFO client.RMProxy: Connecting to ResourceManager at apache/192.168.0.176:8032 19/03/05 09:48:07 INFO mapreduce.JobSubmitter: number of splits:24 19/03/05 09:48:08 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1551343125387_0009 19/03/05 09:48:08 INFO impl.YarnClientImpl: Submitted application application_1551343125387_0009 19/03/05 09:48:08 INFO mapreduce.Job: The url to track the job: http://apache:8088/proxy/application_1551343125387_0009/ 19/03/05 09:48:08 INFO tools.DistCp: DistCp job-id: job_1551343125387_0009 19/03/05 09:48:08 INFO mapreduce.Job: Running job: job_1551343125387_0009 19/03/05 09:48:14 INFO mapreduce.Job: Job job_1551343125387_0009 running in uber mode : false 19/03/05 09:48:14 INFO mapreduce.Job: map 0% reduce 0% 19/03/05 09:48:24 INFO mapreduce.Job: map 4% reduce 0% 19/03/05 09:48:27 INFO mapreduce.Job: map 8% reduce 0% ...... 19/03/05 09:49:18 INFO mapreduce.Job: map 92% reduce 0% 19/03/05 09:49:20 INFO mapreduce.Job: map 96% reduce 0% 19/03/05 09:49:21 INFO mapreduce.Job: map 100% reduce 0% 19/03/05 09:49:22 INFO mapreduce.Job: Job job_1551343125387_0009 completed successfully 19/03/05 09:49:22 INFO mapreduce.Job: Counters: 38 File System Counters FILE: Number of bytes read=0 FILE: Number of bytes written=2932910 FILE: Number of read operations=0 FILE: Number of large read operations=0 FILE: Number of write operations=0 HDFS: Number of bytes read=1099535478 HDFS: Number of bytes written=0 HDFS: Number of read operations=548 HDFS: Number of large read operations=0 HDFS: Number of write operations=48 OSS: Number of bytes read=0 OSS: Number of bytes written=1099511600 OSS: Number of read operations=1262 OSS: Number of large read operations=0 OSS: Number of write operations=405 ...... [root@apache hadoop-2.7.2]# hadoop fs -ls oss://{your-bucket-name}/data/output Found 101 items -rw-rw-rw- 1 root root 0 2019-03-05 09:48 oss://{your-bucket-name}/data/output/_SUCCESS -rw-rw-rw- 1 root root 10995200 2019-03-05 09:48 oss://{your-bucket-name}/data/output/part-m-00000 -rw-rw-rw- 1 root root 10995100 2019-03-05 09:48 oss://{your-bucket-name}/data/output/part-m-00001 ...... 参考链接 https://yq.aliyun.com/articles/292792?spm=a2c4e.11155435.0.0.7ccba82fbDwfhK https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md
-
发表了文章 2019-01-04
使用Logstash将日志导入OSS
前言 logstash 是开源的服务器端数据处理管道，能够同时从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的 “存储库” 中。 目前logstash已经支持了很多输入和输出插件，本文主要介绍如何使用logstash output oss插件将日志导入OSS。 安装 目前logstash output oss插件可以在RubyGems.org里面找到，并且logstash使用RubyGems.org作为它的插件仓库，因此我们可以直接进入到logstash的安装目录，执行如下命令进行安装 ./bin/logstash-plugin install logstash-output-oss 将会产生如下输出 Validating logstash-output-oss Installing logstash-output-oss com.aliyun.oss:aliyun-sdk-oss:3.4.0:compile Installation successful 我们现在可以查看刚才安装的logstash output oss插件 [user@host logstash-6.5.3]# bin/logstash-plugin list --verbose logstash-output-oss logstash-output-oss (0.1.1) 配置及运行 logstash output oss插件的相关配置项的解释可以在这里找到，需要注意的是，endpoint/bucket/access_key_id/access_key_secret都是必填项，其他的都是可选项。下面是一个示例，用来将写入/etc/logstash-6.5.3/sample.data中事件，导入到OSS里面。 input { file { path => "/etc/logstash-6.5.3/sample.data" codec => json { charset => "UTF-8" } } } output { oss { "endpoint" => "oss-cn-zhangjiakou-internal.aliyuncs.com" "bucket" => "Your bucket" "prefix" => "logstash/%{index}" "recover" => true "access_key_id" => "Your access key id" "access_key_secret" => "Your access secret key" "rotation_strategy" => "size_and_time" "time_rotate" => 2 codec => json_lines { charset => "UTF-8" } } } 其中，我们生成的event脚本是: while [[ 1 ]] do let index=$RANDOM%10 echo "{\"index\": $index, \"message\": \"Hello, world\"}" >> sample.data done 我们配置的rotata的策略是size_and_time，基于时间和大小，其中配置了时间是每两分钟(time_rotate)，大小采用的是默认值(在代码里面定义，30MB)。 在这里注意一下配置项中的prefix(logstash/%{index})字段，这个字段支持string interpolation，因此它会用event中相同的字段进行替换(如果没有则不替换)。 最终，我们可以通过控制台来查看生成的文件： 参考资料 https://rubygems.org/gems/logstash-output-osshttps://github.com/aliyun/logstash-output-osshttps://www.elastic.co/guide/en/logstash/current/index.html
-
发表了文章 2018-11-01
Spark读写OSS并使用OSS Select来加速查询
Spark读写OSS 基于这篇文章搭建的CDH6以及配置，我们来使Spark能够读写OSS(其他版本的Spark都是类似的做法，不再赘述)。 由于默认Spark并没有将OSS的支持包放到它的CLASSPATH里面，所以我们需要执行如下命令下面的步骤需要在所有的CDH节点执行 进入到$CDH_HOME/lib/spark目录, 执行如下命令 [root@cdh-master spark]# cd jars/ [root@cdh-master jars]# ln -s ../../../jars/hadoop-aliyun-3.0.0-cdh6.0.1.jar hadoop-aliyun.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-sdk-oss-2.8.3.jar aliyun-sdk-oss-2.8.3.jar [root@cdh-master jars]# ln -s ../../../jars/jdom-1.1.jar jdom-1.1.jar 进入到$CDH_HOME/lib/spark目录，运行一个查询 [root@cdh-master spark]# ./bin/spark-shell WARNING: User-defined SPARK_HOME (/opt/cloudera/parcels/CDH-6.0.1-1.cdh6.0.1.p0.590678/lib/spark) overrides detected (/opt/cloudera/parcels/CDH/lib/spark). WARNING: Running spark-class from user-defined location. Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). Spark context Web UI available at http://x.x.x.x:4040 Spark context available as 'sc' (master = yarn, app id = application_1540878848110_0004). Spark session available as 'spark'. Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 2.2.0-cdh6.0.1 /_/ Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_152) Type in expressions to have them evaluated. Type :help for more information. scala> val myfile = sc.textFile("oss://{your-bucket-name}/50/store_sales") myfile: org.apache.spark.rdd.RDD[String] = oss://{your-bucket-name}/50/store_sales MapPartitionsRDD[1] at textFile at <console>:24 scala> myfile.count() res0: Long = 144004764 scala> myfile.map(line => line.split('|')).filter(_(0).toInt >= 2451262).take(3) res15: Array[Array[String]] = Array(Array(2451262, 71079, 20359, 154660, 284233, 6206, 150579, 46, 512, 2160001, 84, 6.94, 11.38, 9.33, 681.83, 783.72, 582.96, 955.92, 5.09, 681.83, 101.89, 106.98, -481.07), Array(2451262, 71079, 26863, 154660, 284233, 6206, 150579, 46, 345, 2160001, 12, 67.82, 115.29, 25.36, 0.00, 304.32, 813.84, 1383.48, 21.30, 0.00, 304.32, 325.62, -509.52), Array(2451262, 71079, 55852, 154660, 284233, 6206, 150579, 46, 243, 2160001, 74, 32.41, 34.67, 1.38, 0.00, 102.12, 2398.34, 2565.58, 4.08, 0.00, 102.12, 106.20, -2296.22)) scala> myfile.map(line => line.split('|')).filter(_(0) >= "2451262").saveAsTextFile("oss://{your-bucket-name}/spark-oss-test.1") Spark支持OSS Select 这篇文章介绍了OSS Select，OSS Select目前已经在深圳区域实现商业化，下面的实验将基于oss-cn-shenzhen.aliyuncs.com这个OSS EndPoint来进行(基于CDH6，其他版本的Spark做法类似)。 部署 下面的步骤需要在所有的CDH节点执行 下载OSS Select的Spark支持包(目前该支持包还在测试中)，放到$CDH_HOME/jars下 http://gosspublic.alicdn.com/hadoop-spark/spark-2.2.0-oss-select-0.1.0-SNAPSHOT.tar.gz [root@cdh-master jars]# pwd /opt/cloudera/parcels/CDH/jars [root@cdh-master jars]# ls -ltrh ..... -rw-r--r-- 1 root root 20K 9月 20 03:45 jsr305-3.0.1.jar -rw-r--r-- 1 root root 114K 10月 30 16:12 aliyun-java-sdk-core-3.4.0.jar -rw-r--r-- 1 root root 770K 10月 30 16:12 aliyun-java-sdk-ecs-4.2.0.jar -rw-r--r-- 1 root root 535K 10月 30 16:12 aliyun-sdk-oss-3.3.0.jar -rw-r--r-- 1 root root 66K 10月 30 16:12 aliyun-oss-select-spark_2.11-0.1.0-SNAPSHOT.jar -rw-r--r-- 1 root root 13K 10月 30 16:12 aliyun-java-sdk-sts-3.0.0.jar -rw-r--r-- 1 root root 211K 10月 30 16:12 aliyun-java-sdk-ram-3.0.0.jar -rw-r--r-- 1 root root 870K 10月 30 16:13 jaxb-impl-2.2.3-1.jar -rw-r--r-- 1 root root 150K 10月 30 16:13 jdom-1.1.jar -rw-r--r-- 1 root root 145K 10月 30 16:13 jersey-json-1.9.jar -rw-r--r-- 1 root root 448K 10月 30 16:13 jersey-core-1.9.jar -rw-r--r-- 1 root root 56K 10月 30 16:13 json-20170516.jar -rw-r--r-- 1 root root 67K 10月 30 16:13 jettison-1.1.jar -rw-r--r-- 1 root root 26K 10月 30 16:13 stax-api-1.0.1.jar 进入到$CDH_HOME/lib/spark/jars [root@cdh-master jars]# pwd /opt/cloudera/parcels/CDH/lib/spark/jars [root@cdh-master jars]# rm -f aliyun-sdk-oss-2.8.3.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-oss-select-spark_2.11-0.1.0-SNAPSHOT.jar aliyun-oss-select-spark_2.11-0.1.0-SNAPSHOT.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-java-sdk-core-3.4.0.jar aliyun-java-sdk-core-3.4.0.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-java-sdk-ecs-4.2.0.jar aliyun-java-sdk-ecs-4.2.0.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-java-sdk-ram-3.0.0.jar aliyun-java-sdk-ram-3.0.0.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-java-sdk-sts-3.0.0.jar aliyun-java-sdk-sts-3.0.0.jar [root@cdh-master jars]# ln -s ../../../jars/aliyun-sdk-oss-3.3.0.jar aliyun-sdk-oss-3.3.0.jar [root@cdh-master jars]# ln -s ../../../jars/jdom-1.1.jar jdom-1.1.jar 对比测试 这里使用的是spark on yarn，其中Node Manager节点是4个，每个节点最多可以运行4个container，每个container配备的资源是1核2GB内存。测试数据共630MB，包含3列，分别是姓名、公司和年龄。 [root@cdh-master jars]# hadoop fs -ls oss://select-test-sz/people/ Found 10 items -rw-rw-rw- 1 63079930 2018-10-30 17:03 oss://select-test-sz/people/part-00000 -rw-rw-rw- 1 63079930 2018-10-30 17:03 oss://select-test-sz/people/part-00001 -rw-rw-rw- 1 63079930 2018-10-30 17:05 oss://select-test-sz/people/part-00002 -rw-rw-rw- 1 63079930 2018-10-30 17:05 oss://select-test-sz/people/part-00003 -rw-rw-rw- 1 63079930 2018-10-30 17:06 oss://select-test-sz/people/part-00004 -rw-rw-rw- 1 63079930 2018-10-30 17:12 oss://select-test-sz/people/part-00005 -rw-rw-rw- 1 63079930 2018-10-30 17:14 oss://select-test-sz/people/part-00006 -rw-rw-rw- 1 63079930 2018-10-30 17:14 oss://select-test-sz/people/part-00007 -rw-rw-rw- 1 63079930 2018-10-30 17:15 oss://select-test-sz/people/part-00008 -rw-rw-rw- 1 63079930 2018-10-30 17:16 oss://select-test-sz/people/part-00009 进入到$CDH_HOME/lib/spark/，启动spark-shell [root@cdh-master spark]# ./bin/spark-shell WARNING: User-defined SPARK_HOME (/opt/cloudera/parcels/CDH-6.0.1-1.cdh6.0.1.p0.590678/lib/spark) overrides detected (/opt/cloudera/parcels/CDH/lib/spark). WARNING: Running spark-class from user-defined location. Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). Spark context Web UI available at http://x.x.x.x:4040 Spark context available as 'sc' (master = yarn, app id = application_1540887123331_0008). Spark session available as 'spark'. Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 2.2.0-cdh6.0.1 /_/ Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_152) Type in expressions to have them evaluated. Type :help for more information. scala> val sqlContext = spark.sqlContext sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@4bdef487 scala> sqlContext.sql("CREATE TEMPORARY VIEW people USING com.aliyun.oss " + | "OPTIONS (" + | "oss.bucket 'select-test-sz', " + | "oss.prefix 'people', " + // objects with this prefix belong to this table | "oss.schema 'name string, company string, age long'," + // like 'column_a long, column_b string' | "oss.data.format 'csv'," + // we only support csv now | "oss.input.csv.header 'None'," + | "oss.input.csv.recordDelimiter '\r\n'," + | "oss.input.csv.fieldDelimiter ','," + | "oss.input.csv.commentChar '#'," + | "oss.input.csv.quoteChar '\"'," + | "oss.output.csv.recordDelimiter '\n'," + | "oss.output.csv.fieldDelimiter ','," + | "oss.output.csv.quoteChar '\"'," + | "oss.endpoint 'oss-cn-shenzhen.aliyuncs.com', " + | "oss.accessKeyId 'Your Access Key Id', " + | "oss.accessKeySecret 'Your Access Key Secret')") res0: org.apache.spark.sql.DataFrame = [] scala> val sql: String = "select count(*) from people where name like 'Lora%'" sql: String = select count(*) from people where name like 'Lora%' scala> sqlContext.sql(sql).show() +--------+ |count(1)| +--------+ | 31770| +--------+ scala> val textFile = sc.textFile("oss://select-test-sz/people/") textFile: org.apache.spark.rdd.RDD[String] = oss://select-test-sz/people/ MapPartitionsRDD[8] at textFile at <console>:24 scala> textFile.map(line => line.split(',')).filter(_(0).startsWith("Lora")).count() res3: Long = 31770 然后我们分别看使用OSS Select与不使用OSS Select的时间对比，可以看到，使用OSS Select的时间是不使用OSS Select时间的四分之一。 Spark对接OSS Select支持包的实现(Preview) 我们通过扩展Spark的DataSource API来实现Spark对接OSS Select。通过实现PrunedFilteredScan，我们可以把需要的列和过滤条件下推到OSS Select执行。目前这个支持包还在开发中，定义的规范如下： scala> sqlContext.sql("CREATE TEMPORARY VIEW people USING com.aliyun.oss " + | "OPTIONS (" + | "oss.bucket 'select-test-sz', " + | "oss.prefix 'people', " + // objects with this prefix belong to this table | "oss.schema 'name string, company string, age long'," + // like 'column_a long, column_b string' | "oss.data.format 'csv'," + // we only support csv now | "oss.input.csv.header 'None'," + | "oss.input.csv.recordDelimiter '\r\n'," + | "oss.input.csv.fieldDelimiter ','," + | "oss.input.csv.commentChar '#'," + | "oss.input.csv.quoteChar '\"'," + | "oss.output.csv.recordDelimiter '\n'," + | "oss.output.csv.fieldDelimiter ','," + | "oss.output.csv.quoteChar '\"'," + | "oss.endpoint 'oss-cn-shenzhen.aliyuncs.com', " + | "oss.accessKeyId 'Your Access Key Id', " + | "oss.accessKeySecret 'Your Access Key Secret')") 字段 说明 oss.bucket 数据所在的bucket oss.prefix 拥有这个前缀的Object都属于定义的这个TEMPORARY VIEW oss.schema 这个 TEMPORARY VIEW的schema，目前通过字符串指定，后续会通过一个文件来指定schema oss.data.format 数据内容的格式，目前支持CSV格式，其他格式也会陆续支持 oss.input.csv.* 定义CSV输入格式参数 oss.output.csv.* 定义CSV输出格式参数 oss.endpoint bucket所在的Endpoint oss.accessKeyId 你的Access Key Id oss.accessKeySecret 你的Access Key Secret 目前只定义基本参数，可以参考OSS Select API文档，其余的参数也在支持中。 支持的过滤条件：=,<,>,<=, >=,||,or,not,and,in,like(StringStartsWith,StringEndsWith,StringContains)。对于不能下推的过滤条件(如算术运算、字符串拼接等，这些通过PrunedFilteredScan获取不到)，则只下推需要的列到OSS Select。 然而，OSS Select还支持其他过滤条件，可以参考OSS Select API文档 对比TPC-H的查询 主要对比TPC-H中query1.sql对于lineitem这个table的查询性能，为了能使OSS Select过滤更多的数据，我们将where条件改一下(由l_shipdate <= '1998-09-16'改为where l_shipdate > '1997-09-16')，测试数据大小是2.27GB [root@cdh-master ~]# hadoop fs -ls oss://select-test-sz/data/lineitem.csv -rw-rw-rw- 1 2441079322 2018-10-31 11:18 oss://select-test-sz/data/lineitem.csv 对比如下 scala> import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType, DoubleType} import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType, DoubleType} scala> import org.apache.spark.sql.{Row, SQLContext} import org.apache.spark.sql.{Row, SQLContext} scala> val sqlContext = spark.sqlContext sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@74e2cfc5 scala> val textFile = sc.textFile("oss://select-test-sz/data/lineitem.csv") textFile: org.apache.spark.rdd.RDD[String] = oss://select-test-sz/data/lineitem.csv MapPartitionsRDD[1] at textFile at <console>:26 scala> val dataRdd = textFile.map(_.split('|')) dataRdd: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[2] at map at <console>:28 scala> val schema = StructType( | List( | StructField("L_ORDERKEY",LongType,true), | StructField("L_PARTKEY",LongType,true), | StructField("L_SUPPKEY",LongType,true), | StructField("L_LINENUMBER",IntegerType,true), | StructField("L_QUANTITY",DoubleType,true), | StructField("L_EXTENDEDPRICE",DoubleType,true), | StructField("L_DISCOUNT",DoubleType,true), | StructField("L_TAX",DoubleType,true), | StructField("L_RETURNFLAG",StringType,true), | StructField("L_LINESTATUS",StringType,true), | StructField("L_SHIPDATE",StringType,true), | StructField("L_COMMITDATE",StringType,true), | StructField("L_RECEIPTDATE",StringType,true), | StructField("L_SHIPINSTRUCT",StringType,true), | StructField("L_SHIPMODE",StringType,true), | StructField("L_COMMENT",StringType,true) | ) | ) schema: org.apache.spark.sql.types.StructType = StructType(StructField(L_ORDERKEY,LongType,true), StructField(L_PARTKEY,LongType,true), StructField(L_SUPPKEY,LongType,true), StructField(L_LINENUMBER,IntegerType,true), StructField(L_QUANTITY,DoubleType,true), StructField(L_EXTENDEDPRICE,DoubleType,true), StructField(L_DISCOUNT,DoubleType,true), StructField(L_TAX,DoubleType,true), StructField(L_RETURNFLAG,StringType,true), StructField(L_LINESTATUS,StringType,true), StructField(L_SHIPDATE,StringType,true), StructField(L_COMMITDATE,StringType,true), StructField(L_RECEIPTDATE,StringType,true), StructField(L_SHIPINSTRUCT,StringType,true), StructField(L_SHIPMODE,StringType,true), StructField(L_COMMENT,StringType,true)) scala> val dataRowRdd = dataRdd.map(p => Row(p(0).toLong, p(1).toLong, p(2).toLong, p(3).toInt, p(4).toDouble, p(5).toDouble, p(6).toDouble, p(7).toDouble, p(8), p(9), p(10), p(11), p(12), p(13), p(14), p(15))) dataRowRdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[3] at map at <console>:30 scala> val dataFrame = sqlContext.createDataFrame(dataRowRdd, schema) dataFrame: org.apache.spark.sql.DataFrame = [L_ORDERKEY: bigint, L_PARTKEY: bigint ... 14 more fields] scala> dataFrame.createOrReplaceTempView("lineitem") scala> spark.sql("select l_returnflag, l_linestatus, sum(l_quantity) as sum_qty, sum(l_extendedprice) as sum_base_price, sum(l_extendedprice * (1 - l_discount)) as sum_disc_price, sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge, avg(l_quantity) as avg_qty, avg(l_extendedprice) as avg_price, avg(l_discount) as avg_disc, count(*) as count_order from lineitem where l_shipdate > '1997-09-16' group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus").show() +------------+------------+-----------+--------------------+--------------------+--------------------+------------------+------------------+-------------------+-----------+ |l_returnflag|l_linestatus| sum_qty| sum_base_price| sum_disc_price| sum_charge| avg_qty| avg_price| avg_disc|count_order| +------------+------------+-----------+--------------------+--------------------+--------------------+------------------+------------------+-------------------+-----------+ | N| O|7.5697385E7|1.135107538838699...|1.078345555027154...|1.121504616321447...|25.501957856643052|38241.036487881756|0.04999335309103123| 2968297| +------------+------------+-----------+--------------------+--------------------+--------------------+------------------+------------------+-------------------+-----------+ scala> sqlContext.sql("CREATE TEMPORARY VIEW item USING com.aliyun.oss " + | "OPTIONS (" + | "oss.bucket 'select-test-sz', " + | "oss.prefix 'data', " + | "oss.schema 'L_ORDERKEY long, L_PARTKEY long, L_SUPPKEY long, L_LINENUMBER int, L_QUANTITY double, L_EXTENDEDPRICE double, L_DISCOUNT double, L_TAX double, L_RETURNFLAG string, L_LINESTATUS string, L_SHIPDATE string, L_COMMITDATE string, L_RECEIPTDATE string, L_SHIPINSTRUCT string, L_SHIPMODE string, L_COMMENT string'," + | "oss.data.format 'csv'," + // we only support csv now | "oss.input.csv.header 'None'," + | "oss.input.csv.recordDelimiter '\n'," + | "oss.input.csv.fieldDelimiter '|'," + | "oss.input.csv.commentChar '#'," + | "oss.input.csv.quoteChar '\"'," + | "oss.output.csv.recordDelimiter '\n'," + | "oss.output.csv.fieldDelimiter ','," + | "oss.output.csv.commentChar '#'," + | "oss.output.csv.quoteChar '\"'," + | "oss.endpoint 'oss-cn-shenzhen.aliyuncs.com', " + | "oss.accessKeyId 'Your Access Key Id', " + | "oss.accessKeySecret 'Your Access Key Secret')") res2: org.apache.spark.sql.DataFrame = [] scala> sqlContext.sql("select l_returnflag, l_linestatus, sum(l_quantity) as sum_qty, sum(l_extendedprice) as sum_base_price, sum(l_extendedprice * (1 - l_discount)) as sum_disc_price, sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge, avg(l_quantity) as avg_qty, avg(l_extendedprice) as avg_price, avg(l_discount) as avg_disc, count(*) as count_order from item where l_shipdate > '1997-09-16' group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus").show() scala> sqlContext.sql("select l_returnflag, l_linestatus, sum(l_quantity) as sum_qty, sum(l_extendedprice) as sum_base_price, sum(l_extendedprice * (1 - l_discount)) as sum_disc_price, sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge, avg(l_quantity) as avg_qty, avg(l_extendedprice) as avg_price, avg(l_discount) as avg_disc, count(*) as count_order from item where l_shipdate > '1997-09-16' group by l_returnflag, l_linestatus order by l_returnflag, l_linestatus").show() +------------+------------+-----------+--------------------+--------------------+--------------------+------------------+-----------------+-------------------+-----------+ |l_returnflag|l_linestatus| sum_qty| sum_base_price| sum_disc_price| sum_charge| avg_qty| avg_price| avg_disc|count_order| +------------+------------+-----------+--------------------+--------------------+--------------------+------------------+-----------------+-------------------+-----------+ | N| O|7.5697385E7|1.135107538838701E11|1.078345555027154...|1.121504616321447...|25.501957856643052|38241.03648788181|0.04999335309103024| 2968297| +------------+------------+-----------+--------------------+--------------------+--------------------+------------------+-----------------+-------------------+-----------+ 耗时对比如下 其中使用Spark SQL与在Spark SQL上使用OSS Select耗时分别是2.5分钟和38秒。 参考文章 https://yq.aliyun.com/articles/593910https://yq.aliyun.com/articles/659735https://yq.aliyun.com/articles/658473https://mapr.com/blog/spark-data-source-api-extending-our-spark-sql-query-engine/
-
发表了文章 2018-10-30
使用Apache Impala(CDH6)查询OSS的数据
CDH6 目前CDH的最新版本是6.0.1，支持Hadoop 3.0.0，本文将介绍如何使CDH6的相关组件(Hadoop/Hive/Spark/Impala等)能够读写OSS。CDH5对OSS的支持在这篇文章介绍。 CDH6支持读写OSS 搭建CDH集群 首先根据官方文档搭建好CDH6集群 增加OSS配置 通过CM来增加配置(对于没有CM管理的集群，可以通过修改core-site.xml来达到)这里以CM为例，需要增加如下配置: 配置项 值 说明 fs.oss.endpoint 如 oss-cn-zhangjiakou-internal.aliyuncs.com 要连接的endpoint fs.oss.accessKeyId access key id fs.oss.accessKeySecret access key secret fs.oss.impl org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem hadoop oss文件系统实现类，目前固定为这个 fs.oss.buffer.dir /tmp/oss 临时文件目录 fs.oss.connection.secure.enabled false 是否enable https, 根据需要来设置，enable https会影响性能 fs.oss.connection.maximum 2048 与oss的连接数，根据需要设置 相关参数的解释可以在这里找到 重启集群，验证读写OSS 增加配置后，根据CM提示重启集群，重启后，可以测试 # 测试写 hadoop fs -mkdir oss://{your-bucket-name}/hadoop-test # 测试读 hadoop fs -ls oss://{your-bucket-name}/ Apache Impala查询OSS的数据 关于Apache Impala的介绍可以查看官方文档，下面主要介绍使用Apache Impala查询OSS，并且运行TPC-DS的那些查询语句。 虽然CDH6天然支持OSS，但是它里面的Impala组件却默认没有将OSS的支持放到它的CLASSPATH里面去，因此我们需要在所有的Impala节点执行如下命令下面的步骤需要在所有的Impala节点执行 进入到$CDH_HOME/lib/impala目录, 执行如下命令 [root@cdh-master impala]# cd lib/ [root@cdh-master lib]# ln -s ../../../jars/hadoop-aliyun-3.0.0-cdh6.0.1.jar hadoop-aliyun.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-sdk-oss-2.8.3.jar aliyun-sdk-oss-2.8.3.jar [root@cdh-master lib]# ln -s ../../../jars/jdom-1.1.jar jdom-1.1.jar 进入到$CDH_HOME/bin目录，修改impalad/statestored/catalogd这三个文件，在文件最后一行exec命令前，增加如下一行 export CLASSPATH=$CLASSPATH:${IMPALA_HOME}/lib/hadoop-aliyun.jar:${IMPALA_HOME}/lib/aliyun-sdk-oss-2.8.3.jar:${IMPALA_HOME}/lib/jdom-1.1.jar 重启所有节点的impala相关进程，这样impala就可以查询OSS的数据。 运行TPC-DS查询 TPC-DS查询在这里，https://github.com/hortonworks/hive-testbench/tree/hive14基于Hive，因为Hive QL与Impala SQl兼容性比较高，所以我们用这个作为实验(有几个query因为兼容性问题不能运行) 执行如下命令，生成sample数据到OSS [root@cdh-master ~]# git clone https://github.com/hortonworks/hive-testbench.git [root@cdh-master ~]# cd hive-testbench [root@cdh-master hive-testbench]# git checkout hive14 [root@cdh-master hive-testbench]# ./tpcds-build.sh [root@cdh-master hive-testbench]# FORMAT=textfile ./tpcds-setup.sh 50 oss://{your-bucket-name}/ 这个benchmark的建表语句与Apache Impala的建表语句兼容，复制ddl-tpcds/text/alltables.sql中的建表语句，修改${LOCATION}即可，例如： [root@cdh-master hive-testbench]# impala-shell -i cdh-slave01 -d default Starting Impala Shell without Kerberos authentication Connected to cdh-slave01:21000 Server version: impalad version 3.0.0-cdh6.0.1 RELEASE (build 9a74a5053de5f7b8dd983802e6d75e58d31472db) *********************************************************************************** Welcome to the Impala shell. (Impala Shell v3.0.0-cdh6.0.1 (9a74a50) built on Wed Sep 19 11:27:37 PDT 2018) Want to know what version of Impala you're connected to? Run the VERSION command to find out! *********************************************************************************** Query: use `default` [cdh-slave01:21000] default> create external table call_center( > cc_call_center_sk bigint > , cc_call_center_id string > , cc_rec_start_date string > , cc_rec_end_date string > , cc_closed_date_sk bigint > , cc_open_date_sk bigint > , cc_name string > , cc_class string > , cc_employees int > , cc_sq_ft int > , cc_hours string > , cc_manager string > , cc_mkt_id int > , cc_mkt_class string > , cc_mkt_desc string > , cc_market_manager string > , cc_division int > , cc_division_name string > , cc_company int > , cc_company_name string > , cc_street_number string > , cc_street_name string > , cc_street_type string > , cc_suite_number string > , cc_city string > , cc_county string > , cc_state string > , cc_zip string > , cc_country string > , cc_gmt_offset double > , cc_tax_percentage double > ) > row format delimited fields terminated by '|' > location 'oss://{your-bucket-name}/50/call_center'; Query: create external table call_center( cc_call_center_sk bigint , cc_call_center_id string , cc_rec_start_date string , cc_rec_end_date string , cc_closed_date_sk bigint , cc_open_date_sk bigint , cc_name string , cc_class string , cc_employees int , cc_sq_ft int , cc_hours string , cc_manager string , cc_mkt_id int , cc_mkt_class string , cc_mkt_desc string , cc_market_manager string , cc_division int , cc_division_name string , cc_company int , cc_company_name string , cc_street_number string , cc_street_name string , cc_street_type string , cc_suite_number string , cc_city string , cc_county string , cc_state string , cc_zip string , cc_country string , cc_gmt_offset double , cc_tax_percentage double ) row format delimited fields terminated by '|' location 'oss://{your-bucket-name}/50/call_center' +-------------------------+ | summary | +-------------------------+ | Table has been created. | +-------------------------+ Fetched 1 row(s) in 4.10s 注意将建表语句中的{your-bucket-name}替换为你自己的bucket所有的表都建好后，我们可以查看一下 [cdh-slave01:21000] default> show tables; Query: show tables +------------------------+ | name | +------------------------+ | call_center | | catalog_page | | catalog_returns | | catalog_sales | | customer | | customer_address | | customer_demographics | | date_dim | | household_demographics | | income_band | | inventory | | item | | promotion | | reason | | ship_mode | | store | | store_returns | | store_sales | | time_dim | | warehouse | | web_page | | web_returns | | web_sales | | web_site | +------------------------+ Fetched 24 row(s) in 0.03s 下面我们就可以在Impala上执行TPC-DS的查询，查询的SQL文件在sample-queries-tpcds目录下 [root@cdh-master hive-testbench]# ls sample-queries-tpcds query12.sql query20.sql query27.sql query39.sql query46.sql query54.sql query64.sql query71.sql query7.sql query87.sql query93.sql README.md query13.sql query21.sql query28.sql query3.sql query48.sql query55.sql query65.sql query72.sql query80.sql query88.sql query94.sql testbench.settings query15.sql query22.sql query29.sql query40.sql query49.sql query56.sql query66.sql query73.sql query82.sql query89.sql query95.sql testbench-withATS.settings query17.sql query24.sql query31.sql query42.sql query50.sql query58.sql query67.sql query75.sql query83.sql query90.sql query96.sql query18.sql query25.sql query32.sql query43.sql query51.sql query60.sql query68.sql query76.sql query84.sql query91.sql query97.sql query19.sql query26.sql query34.sql query45.sql query52.sql query63.sql query70.sql query79.sql query85.sql query92.sql query98.sql 我们执行query13.sql [root@cdh-master hive-testbench]# impala-shell -i cdh-slave01 -d default -f sample-queries-tpcds/query13.sql Starting Impala Shell without Kerberos authentication Connected to cdh-slave01:21000 Server version: impalad version 3.0.0-cdh6.0.1 RELEASE (build 9a74a5053de5f7b8dd983802e6d75e58d31472db) Query: use `default` Query: select avg(ss_quantity) ,avg(ss_ext_sales_price) ,avg(ss_ext_wholesale_cost) ,sum(ss_ext_wholesale_cost) from store_sales ,store ,customer_demographics ,household_demographics ,customer_address ,date_dim where store.s_store_sk = store_sales.ss_store_sk and store_sales.ss_sold_date_sk = date_dim.d_date_sk and date_dim.d_year = 2001 and((store_sales.ss_hdemo_sk=household_demographics.hd_demo_sk and customer_demographics.cd_demo_sk = store_sales.ss_cdemo_sk and customer_demographics.cd_marital_status = 'M' and customer_demographics.cd_education_status = '4 yr Degree' and store_sales.ss_sales_price between 100.00 and 150.00 and household_demographics.hd_dep_count = 3 )or (store_sales.ss_hdemo_sk=household_demographics.hd_demo_sk and customer_demographics.cd_demo_sk = store_sales.ss_cdemo_sk and customer_demographics.cd_marital_status = 'D' and customer_demographics.cd_education_status = 'Primary' and store_sales.ss_sales_price between 50.00 and 100.00 and household_demographics.hd_dep_count = 1 ) or (store_sales.ss_hdemo_sk=household_demographics.hd_demo_sk and customer_demographics.cd_demo_sk = ss_cdemo_sk and customer_demographics.cd_marital_status = 'U' and customer_demographics.cd_education_status = 'Advanced Degree' and store_sales.ss_sales_price between 150.00 and 200.00 and household_demographics.hd_dep_count = 1 )) and((store_sales.ss_addr_sk = customer_address.ca_address_sk and customer_address.ca_country = 'United States' and customer_address.ca_state in ('KY', 'GA', 'NM') and store_sales.ss_net_profit between 100 and 200 ) or (store_sales.ss_addr_sk = customer_address.ca_address_sk and customer_address.ca_country = 'United States' and customer_address.ca_state in ('MT', 'OR', 'IN') and store_sales.ss_net_profit between 150 and 300 ) or (store_sales.ss_addr_sk = customer_address.ca_address_sk and customer_address.ca_country = 'United States' and customer_address.ca_state in ('WI', 'MO', 'WV') and store_sales.ss_net_profit between 50 and 250 )) Query submitted at: 2018-10-30 11:44:47 (Coordinator: http://cdh-slave01:25000) Query progress can be monitored at: http://cdh-slave01:25000/query_plan?query_id=ff4b3157eddfc3c4:8a31c6500000000 +-------------------+-------------------------+----------------------------+----------------------------+ | avg(ss_quantity) | avg(ss_ext_sales_price) | avg(ss_ext_wholesale_cost) | sum(ss_ext_wholesale_cost) | +-------------------+-------------------------+----------------------------+----------------------------+ | 30.87106918238994 | 2352.642327044025 | 2162.600911949685 | 687707.09 | +-------------------+-------------------------+----------------------------+----------------------------+ Fetched 1 row(s) in 353.16s 这个查询涉及到6张表，store/store_sales/customer_demographics/household_demographics/customer_address/date_dim，分别看一下它们的状态 [cdh-slave01:21000] default> show table STATS store; Query: show table STATS store +-------+--------+---------+--------------+-------------------+--------+-------------------+-------------------------------------------+ | #Rows | #Files | Size | Bytes Cached | Cache Replication | Format | Incremental stats | Location | +-------+--------+---------+--------------+-------------------+--------+-------------------+-------------------------------------------+ | -1 | 1 | 37.56KB | NOT CACHED | NOT CACHED | TEXT | false | oss://{your-bucket-name}/50/store | +-------+--------+---------+--------------+-------------------+--------+-------------------+-------------------------------------------+ Fetched 1 row(s) in 0.01s [cdh-slave01:21000] default> show table STATS store_sales; Query: show table STATS store_sales +-------+--------+---------+--------------+-------------------+--------+-------------------+-------------------------------------------------+ | #Rows | #Files | Size | Bytes Cached | Cache Replication | Format | Incremental stats | Location | +-------+--------+---------+--------------+-------------------+--------+-------------------+-------------------------------------------------+ | -1 | 50 | 18.75GB | NOT CACHED | NOT CACHED | TEXT | false | oss://{your-bucket-name}/50/store_sales | +-------+--------+---------+--------------+-------------------+--------+-------------------+-------------------------------------------------+ Fetched 1 row(s) in 0.01s [cdh-slave01:21000] default> show table STATS customer_demographics; Query: show table STATS customer_demographics +-------+--------+---------+--------------+-------------------+--------+-------------------+-----------------------------------------------------------+ | #Rows | #Files | Size | Bytes Cached | Cache Replication | Format | Incremental stats | Location | +-------+--------+---------+--------------+-------------------+--------+-------------------+-----------------------------------------------------------+ | -1 | 50 | 76.92MB | NOT CACHED | NOT CACHED | TEXT | false | oss://{your-bucket-name}/50/customer_demographics | +-------+--------+---------+--------------+-------------------+--------+-------------------+-----------------------------------------------------------+ Fetched 1 row(s) in 0.01s [cdh-slave01:21000] default> show table STATS household_demographics; Query: show table STATS household_demographics +-------+--------+----------+--------------+-------------------+--------+-------------------+------------------------------------------------------------+ | #Rows | #Files | Size | Bytes Cached | Cache Replication | Format | Incremental stats | Location | +-------+--------+----------+--------------+-------------------+--------+-------------------+------------------------------------------------------------+ | -1 | 1 | 148.10KB | NOT CACHED | NOT CACHED | TEXT | false | oss://{your-bucket-name}/50/household_demographics | +-------+--------+----------+--------------+-------------------+--------+-------------------+------------------------------------------------------------+ Fetched 1 row(s) in 0.01s [cdh-slave01:21000] default> show table STATS customer_address; Query: show table STATS customer_address +-------+--------+---------+--------------+-------------------+--------+-------------------+------------------------------------------------------+ | #Rows | #Files | Size | Bytes Cached | Cache Replication | Format | Incremental stats | Location | +-------+--------+---------+--------------+-------------------+--------+-------------------+------------------------------------------------------+ | -1 | 1 | 40.54MB | NOT CACHED | NOT CACHED | TEXT | false | oss://{your-bucket-name}/50/customer_address | +-------+--------+---------+--------------+-------------------+--------+-------------------+------------------------------------------------------+ Fetched 1 row(s) in 0.01s [cdh-slave01:21000] default> show table STATS date_dim; Query: show table STATS date_dim +-------+--------+--------+--------------+-------------------+--------+-------------------+----------------------------------------------+ | #Rows | #Files | Size | Bytes Cached | Cache Replication | Format | Incremental stats | Location | +-------+--------+--------+--------------+-------------------+--------+-------------------+----------------------------------------------+ | -1 | 1 | 9.84MB | NOT CACHED | NOT CACHED | TEXT | false | oss://{your-bucket-name}/50/date_dim | +-------+--------+--------+--------------+-------------------+--------+-------------------+----------------------------------------------+ Fetched 1 row(s) in 0.01s 参考文章 https://yq.aliyun.com/articles/658473?spm=a2c4e.11155435.0.0.2f8b33125xbe9Hhttps://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.mdhttps://yq.aliyun.com/articles/658471?spm=a2c4e.11155435.0.0.2f8b33125xbe9Hhttps://yq.aliyun.com/articles/292792?spm=a2c4e.11155435.0.0.7ccba82fbDwfhK
-
发表了文章 2018-10-29
CDH5 Hadoop如何支持读写OSS
CDH和CM(Cloudera Manager) CDH (Cloudera’s Distribution, including Apache Hadoop)是众多Hadoop发行版本中的一种，由Cloudera维护，目前有不少用户使用这个发行版本。CM是管理集群的工具，使用它可以很方便地部署、安装、运维包括Hadoop、Spark、Hive在内的大数据开源组件。 目前，CDH的最新发布的版本是CDH6.0.1，这个发行版本中，Hadoop的版本是3.0.0，天然支持OSS；而CDH5中的Hadoop是2.6，还不支持OSS(目前Apache Hadoop支持OSS的最低版本是2.9.1)。下面将介绍如何使CDH5支持OSS读写。 CDH5支持读写OSS 问题 由于CDH5的httpclient和httpcore这两个组件版本较低(4.2.5)，而OSS SDK要求这两个组件的版本较高。然而Resource Manager要求的httpclient和httpcore必须是低版本。因此，下面提供了一个workaround方案。 搭建CDH集群 首先，我们根据官方文档搭建好CDH集群(这里以CDH 5.14.4为例)； 下面的步骤需要在所有的CDH节点执行 下载支持包 http://gosspublic.alicdn.com/hadoop-spark/hadoop-oss-cdh-5.14.4.tar.gz 下载这个版本的CDH支持OSS的支持包并解压，里面的文件是： [root@cdh-master ~]# ls -lh hadoop-oss-cdh-5.14.4/ 总用量 2.7M -rw-r--r-- 1 root root 114K 10月 8 17:36 aliyun-java-sdk-core-3.4.0.jar -rw-r--r-- 1 root root 770K 10月 8 17:36 aliyun-java-sdk-ecs-4.2.0.jar -rw-r--r-- 1 root root 211K 10月 8 17:36 aliyun-java-sdk-ram-3.0.0.jar -rw-r--r-- 1 root root 13K 10月 8 17:36 aliyun-java-sdk-sts-3.0.0.jar -rw-r--r-- 1 root root 550K 10月 8 17:36 aliyun-sdk-oss-3.4.1.jar -rw-r--r-- 1 root root 70K 10月 8 17:36 hadoop-aliyun-2.6.0-cdh5.14.4.jar -rw-r--r-- 1 root root 720K 10月 8 18:16 httpclient-4.5.2.jar -rw-r--r-- 1 root root 320K 10月 8 18:16 httpcore-4.4.4.jar 这个支持包是根据CDH5.14.4中Hadoop的版本，并打了Apache Hadoop对OSS支持的patch后编译得到，其他CDH5的小版本对OSS的支持后续也将陆续提供。 部署 将解压后的文件复制到CDH5的安装目录中的jars文件夹中，CDH5的安装目录CDH_HOME(以在/opt/cloudera/parcels/CDH-5.14.4-1.cdh5.14.4.p0.3为例)结构是 [root@cdh-master CDH-5.14.4-1.cdh5.14.4.p0.3]# ls -lh 总用量 100K drwxr-xr-x 2 root root 4.0K 6月 12 21:03 bin drwxr-xr-x 27 root root 4.0K 6月 12 20:57 etc drwxr-xr-x 5 root root 4.0K 6月 12 20:57 include drwxr-xr-x 2 root root 68K 6月 12 21:09 jars drwxr-xr-x 38 root root 4.0K 6月 12 21:03 lib drwxr-xr-x 3 root root 4.0K 6月 12 20:57 lib64 drwxr-xr-x 3 root root 4.0K 6月 12 20:51 libexec drwxr-xr-x 2 root root 4.0K 6月 12 21:02 meta drwxr-xr-x 4 root root 4.0K 6月 12 21:03 share [root@cdh-master CDH-5.14.4-1.cdh5.14.4.p0.3]# ls -ltrh jars/ 总用量 1.4G ....... -rw-r--r-- 1 root root 114K 10月 28 14:27 aliyun-java-sdk-core-3.4.0.jar -rw-r--r-- 1 root root 770K 10月 28 14:27 aliyun-java-sdk-ecs-4.2.0.jar -rw-r--r-- 1 root root 13K 10月 28 14:27 aliyun-java-sdk-sts-3.0.0.jar -rw-r--r-- 1 root root 211K 10月 28 14:27 aliyun-java-sdk-ram-3.0.0.jar -rw-r--r-- 1 root root 550K 10月 28 14:27 aliyun-sdk-oss-3.4.1.jar -rw-r--r-- 1 root root 70K 10月 28 14:27 hadoop-aliyun-2.6.0-cdh5.14.4.jar -rw-r--r-- 1 root root 720K 10月 28 14:27 httpclient-4.5.2.jar -rw-r--r-- 1 root root 320K 10月 28 14:27 httpcore-4.4.4.jar 进入到$CDH_HOME/lib/hadoop目录, 执行如下命令 ```sh [root@cdh-master hadoop]# rm -f lib/httpclient-4.2.5.jar [root@cdh-master hadoop]# rm -f lib/httpcore-4.2.5.jar [root@cdh-master hadoop]# ln -s ../../jars/hadoop-aliyun-2.6.0-cdh5.14.4.jar hadoop-aliyun-2.6.0-cdh5.14.4.jar [root@cdh-master hadoop]# ln -s hadoop-aliyun-2.6.0-cdh5.14.4.jar hadoop-aliyun.jar [root@cdh-master hadoop]# cd lib [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-core-3.4.0.jar aliyun-java-sdk-core-3.4.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-ecs-4.2.0.jar aliyun-java-sdk-ecs-4.2.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-ram-3.0.0.jar aliyun-java-sdk-ram-3.0.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-sts-3.0.0.jar aliyun-java-sdk-sts-3.0.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-sdk-oss-3.4.1.jar aliyun-sdk-oss-3.4.1.jar [root@cdh-master lib]# ln -s ../../../jars/httpclient-4.5.2.jar httpclient-4.5.2.jar [root@cdh-master lib]# ln -s ../../../jars/httpcore-4.4.4.jar httpcore-4.4.4.jar [root@cdh-master lib]# ln -s ../../../jars/jdom-1.1.jar jdom-1.1.jar ``` 进入到Resurce Manager部署节点的$CDH_HOME/lib/hadoop-yarn/bin/目录，将yarn文件中的 CLASSPATH=${CLASSPATH}:$HADOOP_YARN_HOME/${YARN_DIR}/* CLASSPATH=${CLASSPATH}:$HADOOP_YARN_HOME/${YARN_LIB_JARS_DIR}/* 替换为 CLASSPATH=$HADOOP_YARN_HOME/${YARN_DIR}/*:${CLASSPATH} CLASSPATH=$HADOOP_YARN_HOME/${YARN_LIB_JARS_DIR}/*:${CLASSPATH} 进入到Resurce Manager部署节点的$CDH_HOME/lib/hadoop-yarn/lib目录，执行 ```sh [root@cdh-master lib]# ln -s ../../../jars/httpclient-4.2.5.jar httpclient-4.2.5.jar [root@cdh-master lib]# ln -s ../../../jars/httpcore-4.2.5.jar httpcore-4.2.5.jar ``` ###增加OSS配置 通过CM来增加配置(对于没有CM管理的集群，可以通过修改core-site.xml来达到) 这里以CM为例，需要增加如下配置:  配置项 | 值 | 说明 ------- | ------- | ------- fs.oss.endpoint | 如 oss-cn-zhangjiakou-internal.aliyuncs.com | 要连接的endpoint fs.oss.accessKeyId | | access key id fs.oss.accessKeySecret | | access key secret fs.oss.impl | org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem | hadoop oss文件系统实现类，目前固定为这个 fs.oss.buffer.dir | /tmp/oss | 临时文件目录 fs.oss.connection.secure.enabled | false | 是否enable https, 根据需要来设置，enable https会影响性能 fs.oss.connection.maximum | 2048 | 与oss的连接数，根据需要设置 相关参数的解释可以在[**这里**](https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md)找到 ##重启集群，验证读写OSS 增加配置后，根据CM提示重启集群，重启后，可以测试 ```sh # 测试写 hadoop fs -mkdir oss://{your-bucket-name}/hadoop-test # 测试读 hadoop fs -ls oss://{your-bucket-name}/ ``` ## Impala查询OSS中的数据 Impala可以直接查询存储在HDFS的数据，在CDH5支持OSS后，就可以直接查询存储在OSS的数据，由于上面所说的httpclient与httpcore的版本问题，**需要在所有部署impala的节点上**做如下步骤： 进入到$CDH_HOME/lib/impala/lib，执行如下命令 [root@cdh-master lib]# rm -f httpclient-4.2.5.jar httpcore-4.2.5.jar [root@cdh-master lib]# ln -s ../../../jars/httpclient-4.5.2.jar httpclient-4.5.2.jar [root@cdh-master lib]# ln -s ../../../jars/httpcore-4.4.4.jar httpcore-4.4.4.jar [root@cdh-master lib]# ln -s ../../../jars/hadoop-aliyun-2.6.0-cdh5.14.4.jar hadoop-aliyun.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-core-3.4.0.jar aliyun-java-sdk-core-3.4.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-ecs-4.2.0.jar aliyun-java-sdk-ecs-4.2.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-ram-3.0.0.jar aliyun-java-sdk-ram-3.0.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-java-sdk-sts-3.0.0.jar aliyun-java-sdk-sts-3.0.0.jar [root@cdh-master lib]# ln -s ../../../jars/aliyun-sdk-oss-3.4.1.jar aliyun-sdk-oss-3.4.1.jar [root@cdh-master lib]# ln -s ../../../jars/jdom-1.1.jar jdom-1.1.jar 进入到$CDH_HOME/bin目录，修改impalad/statestored/catalogd这三个文件，**在文件最后一行exec命令前，增加如下一行** ``` export CLASSPATH=$CLASSPATH:${IMPALA_HOME}/lib/httpclient-4.5.2.jar:${IMPALA_HOME}/lib/httpcore-4.4.4.jar:${IMPALA_HOME}/lib/hadoop-aliyun.jar:${IMPALA_HOME}/lib/aliyun-java-sdk-core-3.4.0.jar:${IMPALA_HOME}/lib/aliyun-java-sdk-ecs-4.2.0.jar:${IMPALA_HOME}/lib/aliyun-java-sdk-ram-3.0.0.jar:${IMPALA_HOME}/lib/aliyun-java-sdk-sts-3.0.0.jar:${IMPALA_HOME}/lib/aliyun-sdk-oss-3.4.1.jar:${IMPALA_HOME}/lib/jdom-1.1.jar 重启所有节点的impala相关进程，这样impala就可以查询OSS的数据。 示例：TPC-DS的benchmark有一张表为call_center，目前假设存储在OSS中，为此，我们可以创建一个外部表指向它，并且查询这张表根据cc_country分组分别有多少条记录 ```sql [root@cdh-master ~]# impala-shell -i cdh-slave01:21000 Starting Impala Shell without Kerberos authentication Connected to cdh-slave01:21000 Server version: impalad version 2.11.0-cdh5.14.4 RELEASE (build 20e635646a13347800fad36a7d0b1da25ab32404) *********************************************************************************** Welcome to the Impala shell. (Impala Shell v2.11.0-cdh5.14.4 (20e6356) built on Tue Jun 12 03:43:08 PDT 2018) The HISTORY command lists all shell commands in chronological order. *********************************************************************************** [cdh-slave01:21000] > drop table if exists call_center; Query: drop table if exists call_center [cdh-slave01:21000] > [cdh-slave01:21000] > create external table call_center( > cc_call_center_sk bigint > , cc_call_center_id string > , cc_rec_start_date string > , cc_rec_end_date string > , cc_closed_date_sk bigint > , cc_open_date_sk bigint > , cc_name string > , cc_class string > , cc_employees int > , cc_sq_ft int > , cc_hours string > , cc_manager string > , cc_mkt_id int > , cc_mkt_class string > , cc_mkt_desc string > , cc_market_manager string > , cc_division int > , cc_division_name string > , cc_company int > , cc_company_name string > , cc_street_number string > , cc_street_name string > , cc_street_type string > , cc_suite_number string > , cc_city string > , cc_county string > , cc_state string > , cc_zip string > , cc_country string > , cc_gmt_offset double > , cc_tax_percentage double > ) > row format delimited fields terminated by '|' > location 'oss://{bucket}/call_center'; Query: create external table call_center( cc_call_center_sk bigint , cc_call_center_id string , cc_rec_start_date string , cc_rec_end_date string , cc_closed_date_sk bigint , cc_open_date_sk bigint , cc_name string , cc_class string , cc_employees int , cc_sq_ft int , cc_hours string , cc_manager string , cc_mkt_id int , cc_mkt_class string , cc_mkt_desc string , cc_market_manager string , cc_division int , cc_division_name string , cc_company int , cc_company_name string , cc_street_number string , cc_street_name string , cc_street_type string , cc_suite_number string , cc_city string , cc_county string , cc_state string , cc_zip string , cc_country string , cc_gmt_offset double , cc_tax_percentage double ) row format delimited fields terminated by '|' location 'oss://{bucket}/call_center' Fetched 0 row(s) in 0.07s [cdh-slave01:21000] > select cc_country, count(*) from call_center group by cc_country; Query: select cc_country, count(*) from call_center group by cc_country Query submitted at: 2018-10-28 16:21:13 (Coordinator: http://cdh-slave01:25000) Query progress can be monitored at: http://cdh-slave01:25000/query_plan?query_id=fb4e09977145f367:3bdfe4d600000000 +---------------+----------+ | cc_country | count(*) | +---------------+----------+ | United States | 30 | +---------------+----------+ Fetched 1 row(s) in 4.71s 其他版本的支持包 CDH5.8.5 http://gosspublic.alicdn.com/hadoop-spark/hadoop-oss-cdh-5.8.5.tar.gz CDH5.4.4 http://gosspublic.alicdn.com/hadoop-spark/hadoop-oss-cdh-5.4.4.tar.gz CDH6.3.2 http://gosspublic.alicdn.com/hadoop-spark/hadoop-oss-cdh-6.3.2.tar.gz 注：CDH6目前已经支持OSS，但CDH6支持的版本过低，很多优化没有包含在里面，这个支持包会及时包含OSS的优化。将优化包的内容复制到CDH6的安装目录中的jars文件夹中，然后参考CDH5的部署步骤执行即可(主要是更新aliyun-sdk-oss-3.4.1.jar以及将aliyun-java-sdk-*.jar符号链接到对应的位置)。 参考文章 https://yq.aliyun.com/articles/292792?spm=a2c4e.11155435.0.0.7ccba82fbDwfhKhttps://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md
-
发表了文章 2018-10-29
HDP2.6 Hadoop如何支持读写OSS
HDP和Ambari HDP(Hortonworks Data Platform)是由Hortonworks发行的大数据平台，里面包含了Hadoop、Hive、HBase等很多开源组件，目前有不少用户直接使用HDP版本的Hadoop。Ambari是一个分布式工具，可以安装、管理，监控HDP平台。HDP与Ambari的关系，可以类比CDH与CM的关系。目前，HDP的最新版本是3.0.1，里面的Hadoop版本是3.1.1，天然支持了OSS。本文主要介绍如何使低版本的HDP(以HDP2.6.1.0为例)支持读写OSS。 HDP2.6.1.0支持读写OSS HDP2.6.1.0中，Hadoop的版本是2.7.3，还不支持OSS(目前Apache Hadoop支持OSS的最低版本是2.9.1)。 搭建HDP集群 通过官方文档，利用Ambari搭建HDP2.6.1.0的集群(也可以不用Ambari，自行搭建)。 下面的步骤需要在所有的HDP节点执行 下载支持包 http://gosspublic.alicdn.com/hadoop-spark/hadoop-oss-hdp-2.6.1.0-129.tar.gz 下载这个版本的HDP支持OSS的支持包并解压，里面的文件是： [root@hdp-master ~]# ls -lh hadoop-oss-hdp-2.6.1.0-129 total 1.9M -rw-r--r-- 1 501 games 114K Sep 26 15:31 aliyun-java-sdk-core-3.4.0.jar -rw-r--r-- 1 501 games 770K Sep 26 15:31 aliyun-java-sdk-ecs-4.2.0.jar -rw-r--r-- 1 501 games 211K Sep 26 15:31 aliyun-java-sdk-ram-3.0.0.jar -rw-r--r-- 1 501 games 13K Sep 26 15:31 aliyun-java-sdk-sts-3.0.0.jar -rw-r--r-- 1 501 games 550K Sep 26 15:31 aliyun-sdk-oss-3.4.1.jar -rw-r--r-- 1 501 games 67K Sep 26 15:30 hadoop-aliyun-2.7.3.2.6.1.0-129.jar -rw-r--r-- 1 501 games 150K Sep 26 15:31 jdom-1.1.jar 这个支持包是根据HDP2.6.1.0中Hadoop的版本，并打了Apache Hadoop对OSS支持的patch后编译得到，其他HDP2的小版本对OSS的支持后续也将陆续提供。 部署 解压缩后，将hadoop-aliyun-2.7.3.2.6.1.0-129.jar放到/usr/hdp/current/hadoop-client/目录里面；其余的jar文件放到/usr/hdp/current/hadoop-client/lib/里面。其中/usr/hdp/current是当前HDP的安装路径，根据部署环境的不同而不同，用户自己可以做调整放好之后,目录结构如下所示 [root@hdp-master ~]# ls -lh /usr/hdp/current/hadoop-client/hadoop-aliyun-2.7.3.2.6.1.0-129.jar -rw-r--r-- 1 root root 67K Oct 28 20:56 /usr/hdp/current/hadoop-client/hadoop-aliyun-2.7.3.2.6.1.0-129.jar [root@hdp-master ~]# ls -ltrh /usr/hdp/current/hadoop-client/lib total 27M ...... drwxr-xr-x 2 root root 4.0K Oct 28 20:10 ranger-hdfs-plugin-impl drwxr-xr-x 2 root root 4.0K Oct 28 20:10 ranger-yarn-plugin-impl drwxr-xr-x 2 root root 4.0K Oct 28 20:10 native -rw-r--r-- 1 root root 114K Oct 28 20:56 aliyun-java-sdk-core-3.4.0.jar -rw-r--r-- 1 root root 550K Oct 28 20:56 aliyun-sdk-oss-3.4.1.jar -rw-r--r-- 1 root root 13K Oct 28 20:56 aliyun-java-sdk-sts-3.0.0.jar -rw-r--r-- 1 root root 211K Oct 28 20:56 aliyun-java-sdk-ram-3.0.0.jar -rw-r--r-- 1 root root 770K Oct 28 20:56 aliyun-java-sdk-ecs-4.2.0.jar -rw-r--r-- 1 root root 150K Oct 28 20:56 jdom-1.1.jar 增加OSS配置 通过Ambari来增加配置(对于没有用Ambari管理的集群，可以通过修改core-site.xml来达到)这里以Ambari为例，需要增加如下配置: 配置项 值 说明 fs.oss.endpoint 如 oss-cn-zhangjiakou-internal.aliyuncs.com 要连接的endpoint fs.oss.accessKeyId access key id fs.oss.accessKeySecret access key secret fs.oss.impl org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem hadoop oss文件系统实现类，目前固定为这个 fs.oss.buffer.dir /tmp/oss 临时文件目录 fs.oss.connection.secure.enabled false 是否enable https, 根据需要来设置，enable https会影响性能 fs.oss.connection.maximum 2048 与oss的连接数，根据需要设置 相关参数的解释可以在这里找到 重启集群，验证读写OSS 增加配置后，根据Ambari提示重启集群，重启后，可以测试 # 测试写 hadoop fs -mkdir oss://{your-bucket-name}/hadoop-test # 测试读 hadoop fs -ls oss://{your-bucket-name}/ 为了能够运行MAPREDUCE任务，还需要更改hdfs://hdp-master:8020/hdp/apps/2.6.1.0-129/mapreduce/mapreduce.tar.gz包的内容(如果是TEZ类型的作业，则修改hdfs://hdp-master:8020/hdp/apps/2.6.1.0-129/tez/tez.tar.gz即可，其他类型的作业以此类推)，将OSS的支持放进去，执行如下命令 [root@hdp-master ~]# sudo su hdfs [hdfs@hdp-master root]$ cd [hdfs@hdp-master ~]$ hadoop fs -copyToLocal /hdp/apps/2.6.1.0-129/mapreduce/mapreduce.tar.gz . [hdfs@hdp-master ~]$ hadoop fs -rm /hdp/apps/2.6.1.0-129/mapreduce/mapreduce.tar.gz [hdfs@hdp-master ~]$ cp mapreduce.tar.gz mapreduce.tar.gz.bak [hdfs@hdp-master ~]$ tar zxf mapreduce.tar.gz [hdfs@hdp-master ~]$ cp /usr/hdp/current/hadoop-client/hadoop-aliyun-2.7.3.2.6.1.0-129.jar hadoop/share/hadoop/tools/lib/ [hdfs@hdp-master ~]$ cp /usr/hdp/current/hadoop-client/lib/aliyun-* hadoop/share/hadoop/tools/lib/ [hdfs@hdp-master ~]$ cp /usr/hdp/current/hadoop-client/lib/jdom-1.1.jar hadoop/share/hadoop/tools/lib/ [hdfs@hdp-master ~]$ tar zcf mapreduce.tar.gz hadoop [hdfs@hdp-master ~]$ hadoop fs -copyFromLocal mapreduce.tar.gz /hdp/apps/2.6.1.0-129/mapreduce/ 测试teragen和terasort [hdfs@hdp-master ~]$ hadoop jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-mapreduce-examples.jar teragen -Dmapred.map.tasks=100 10995116 oss://{bucket-name}/1G-input 18/10/28 21:32:38 INFO client.RMProxy: Connecting to ResourceManager at cdh-master/192.168.0.161:8050 18/10/28 21:32:38 INFO client.AHSProxy: Connecting to Application History server at cdh-master/192.168.0.161:10200 18/10/28 21:32:38 INFO aliyun.oss: [Server]Unable to execute HTTP request: Not Found [ErrorCode]: NoSuchKey [RequestId]: 5BD5BA7641FCE369BC1D052C [HostId]: null 18/10/28 21:32:38 INFO aliyun.oss: [Server]Unable to execute HTTP request: Not Found [ErrorCode]: NoSuchKey [RequestId]: 5BD5BA7641FCE369BC1D052F [HostId]: null 18/10/28 21:32:39 INFO terasort.TeraSort: Generating 10995116 using 100 18/10/28 21:32:39 INFO mapreduce.JobSubmitter: number of splits:100 18/10/28 21:32:39 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1540728986531_0005 18/10/28 21:32:39 INFO impl.YarnClientImpl: Submitted application application_1540728986531_0005 18/10/28 21:32:39 INFO mapreduce.Job: The url to track the job: http://cdh-master:8088/proxy/application_1540728986531_0005/ 18/10/28 21:32:39 INFO mapreduce.Job: Running job: job_1540728986531_0005 18/10/28 21:32:49 INFO mapreduce.Job: Job job_1540728986531_0005 running in uber mode : false 18/10/28 21:32:49 INFO mapreduce.Job: map 0% reduce 0% 18/10/28 21:32:55 INFO mapreduce.Job: map 1% reduce 0% 18/10/28 21:32:57 INFO mapreduce.Job: map 2% reduce 0% 18/10/28 21:32:58 INFO mapreduce.Job: map 4% reduce 0% ... 18/10/28 21:34:40 INFO mapreduce.Job: map 99% reduce 0% 18/10/28 21:34:42 INFO mapreduce.Job: map 100% reduce 0% 18/10/28 21:35:15 INFO mapreduce.Job: Job job_1540728986531_0005 completed successfully 18/10/28 21:35:15 INFO mapreduce.Job: Counters: 36 ... [hdfs@hdp-master ~]$ hadoop jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-mapreduce-examples.jar terasort -Dmapred.map.tasks=100 oss://{bucket-name}/1G-input oss://{bucket-name}/1G-output 18/10/28 21:39:00 INFO terasort.TeraSort: starting ... 18/10/28 21:39:02 INFO mapreduce.JobSubmitter: number of splits:100 18/10/28 21:39:02 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1540728986531_0006 18/10/28 21:39:02 INFO impl.YarnClientImpl: Submitted application application_1540728986531_0006 18/10/28 21:39:02 INFO mapreduce.Job: The url to track the job: http://cdh-master:8088/proxy/application_1540728986531_0006/ 18/10/28 21:39:02 INFO mapreduce.Job: Running job: job_1540728986531_0006 18/10/28 21:39:09 INFO mapreduce.Job: Job job_1540728986531_0006 running in uber mode : false 18/10/28 21:39:09 INFO mapreduce.Job: map 0% reduce 0% 18/10/28 21:39:17 INFO mapreduce.Job: map 1% reduce 0% 18/10/28 21:39:19 INFO mapreduce.Job: map 2% reduce 0% 18/10/28 21:39:20 INFO mapreduce.Job: map 3% reduce 0% ... 18/10/28 21:42:50 INFO mapreduce.Job: map 100% reduce 75% 18/10/28 21:42:53 INFO mapreduce.Job: map 100% reduce 80% 18/10/28 21:42:56 INFO mapreduce.Job: map 100% reduce 86% 18/10/28 21:42:59 INFO mapreduce.Job: map 100% reduce 92% 18/10/28 21:43:02 INFO mapreduce.Job: map 100% reduce 98% 18/10/28 21:43:05 INFO mapreduce.Job: map 100% reduce 100% ^@18/10/28 21:43:56 INFO mapreduce.Job: Job job_1540728986531_0006 completed successfully 18/10/28 21:43:56 INFO mapreduce.Job: Counters: 54 ... 参考文章 https://yq.aliyun.com/articles/292792?spm=a2c4e.11155435.0.0.7ccba82fbDwfhKhttps://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md
-
发表了文章 2018-09-18
在控制台使用OSS Select
OSS Select-让提取数据价值更简单、更快速 对象存储OSS（Object Storage Service）具有海量、可靠、安全、高性能、低成本的特点。OSS提供标准、低频、归档类型，覆盖多种数据从热到冷的存储需求，单个文件的大小从1字节到48.8TB，可以存储的文件个数无限制。OSS已成为互联网、企业级数据应用的基础设施。通常，获取对象存储数据的通方式为：获取整个对象，或按指定的字节范围来获取数据。OSS Select推出后，可以直接使用简单的SQL语句，从OSS的文件中选取所需要的内容。 本文将介绍如何在控制台轻松体验OSS Select的功能。 准备数据 控制台 登录控制台，并创建bucket 请创建华南1的测试bucket(您可以选择你自己想要创建的bucket名字) 下载模拟数据到本地（该数据本次实验的模拟数据） http://testdatasample.oss-cn-hangzhou.aliyuncs.com/people/people.zip 将下载的数据解压，并放到刚才创建的bucket里 执行SQL，选取内容 点击刚上传Object最右边的更多按钮，然后再点击选取内容按钮，打开选取内容对话框 点击显示文件预览按钮，我们可以预览这个文件我们可以看到这个文件有3列，分别是姓名、公司和年龄。 然后点击下一步按钮，我们可以输入SQL来执行, 比如我们想查找年龄大于50岁，并且名字以Lora开头的人(其中_1, _2, _3是列索引，代表第一列、第二列、第三列以此类推) select * from ossobject where _1 like 'Lora*' and _3 > 50 得到如下结果： 比如我们想统计这个文件有多少行，最大年龄与最小年龄是多少，我们可以执行如下SQL select count(*), max(cast(_3 as int)), min(cast(_3 as int)) from ossobject 得到如下结果： 其他 OSS Select更高级的用法，请参考OSS相关的SDK(Java/Python，其他语言的OSS Select支持在开发中) 参考文章 https://yq.aliyun.com/articles/593910?spm=5176.10695662.1996646101.searchclickresult.41a75825WS8Atv
-
发表了文章 2018-09-06
OSS访问控制介绍
摘要本文主要介绍 OSS 安全管理中访问控制相关的基本概念、机制和使用方法。 背景 OSS作为阿里云的对象存储产品具有快速数据存取无限水平扩展数据容灾等优点可满足海量数据存储需求已服务大量的用户。和任何数据存储一样安全是一个关键特性。用户通过HTTP调用OSS提供认证需要的密钥。因为数据可能通过公共互联网传输这些密钥就是身份凭证而非普通的用户/密码组合。OSS利用这些信息进行用户身份认证及鉴权等访问控制操作。除了以上基于身份的资源管理维度OSS还会考虑Bucket权限维度将两者结合起来构建完整的访问控制体系。 基本概念 OSS权限 OSS使用两层模型管理用户数据object作为基层数据单元与用户的数据对应bucket桶作为管理单元管理objectbucket下所有的object都属于bucket owner由此衍生出了两层的权限bucket acl和object acl。目前在OSS中bucket acl的可选项有private、public-read和public-read-writeobject acl在bucket acl基础上增加default。暂时没有对某个人或者某个组的授权概念。1⃣ bucket acl 权限值 中文名称 权限对访问者的限制 public-read-write 公共读写 任何人包括匿名访问都可以对该Bucket中的Object进行读/写/删除操作所有这些操作产生的费用由该Bucket的Owner承担请慎用该权限。 public-read 公共读私有写 只有该Bucket的Owner或者授权对象可以对存放在其中的Object进行写/删除操作任何人包括匿名访问可以对Object进行读操作。 authenticated-read 授权读私有写 只有该Bucket的Owner或者授权对象可以对存放在其中的Object进行写/删除操作只有通过阿里云身份认证用户可以对Object进行读操作。禁止匿名读 private 私有读写 只有该Bucket的Owner或者授权对象可以对存放在其中的Object进行读/写/删除操作其他人在未经授权的情况下无法访问该Bucket内的Object。 2⃣ object acl 权限值 中文名称 权限对访问者的限制 public-read-write 公共读写 该ACL表明某个Object是公共读写资源即所有用户拥有对该Object的读写权限。 public-read 公共读私有写 该ACL表明某个Object是公共读资源即非Object Owner只有该Object的读权限而Object Owner拥有该Object的读写权限。 authenticated-read 授权读私有写 该ACL表明某个Object是被阿里云认证身份的读资源即禁止匿名用户拥有该Object的读权限而Object Owner拥有该Object的读写权限。 private 私有读写 该ACL表明某个Object是私有资源即只有该Object的Owner拥有该Object的读写权限其他的用户没有权限操作该Object。 default 默认权限 该ACL表明某个Object是遵循Bucket读写权限的资源即Bucket是什么权限Object就是什么权限。 RAM RAM (Resource Access Management)是阿里云为客户提供的用户身份管理与访问控制服务。使用RAM用户可以创建、管理用户账号比如员工、系统或应用程序并可以控制这些用户账号对资源的操作权限。当企业存在多用户协同操作资源时RAM可以按需为用户分配最小权限从而避免与其他用户共享云账号密钥降低企业的信息安全风险。RAM基于身份进行权限管理目前支持用户、组和角色三种身份身份是管理的核心身份需要赋予相应的授权策略所有的行为都是基于特定的身份来展开。 授权策略 授权策略可以分为RamPolicy和BucketPolicy两种包含以下基本元素版本version描述授权策略的版本信息。语句statement描述权限的详细信息。该元素包括授权类型、操作、资源、条件等多个其他元素的权限或权限集合。一条策略可以有一个或多个语句元素。委托人principal描述策略授权的实体。例如用户云账户、ram子账号、用户组等。授权类型effect描述声明产生的结果是“允许”还是“显式拒绝”取值为 Allow 或 Deny。该元素是必填项。操作名称列表action描述允许或拒绝的操作支持多值取值为云服务所定义的API操作名称。该元素是必填项。资源resource描述授权的具体数据。每款产品的资源定义详情会有所区别。该元素是必填项。条件condition描述策略生效的约束条件。条件包括操作符、操作键和操作值组成。条件值可包括时间、IP 地址等信息。有些服务允许指定其他值。该元素是非必填项。 在OSS控制台授权页面可以进行BucketPolicy的图形化操作在访问控制控制台可以进行RamPolicy的图形化操作。 权限体系 访问者身份 用户使用OSS提供的RESTful API或者SDK开发包访问OSS服务每一次请求根据当前Bucket权限和操作不同要求用户进行身份验证或者直接匿名访问。对OSS的资源访问的分类如下 按访问者的角色可分为拥有者访问、RAM用户和STS用户访问。这里的拥有者指的是Bucket的Owner。RAM用户是通过RAM创建的子用户属于云账号不拥有实际的任何资源按照RAM定义的规则去访问Bucket里的资源。STS用户是通过临时访问凭证服务STS颁发的临时用户在一定的生命周期内按照STS定义的规则去访问Bucket里的资源。 按访问者的身份信息可分为匿名访问和带签名访问。对于OSS来说如果请求中没有携带任何和身份相关的信息即为匿名访问。带签名访问指的是按照OSS API文档中规定的在请求头部或者在请求URL中携带签名的相关信息。 请求鉴权 权限管理是管理身份对资源的访问按照权限管理的划分oss将权限验证流程划分为基于身份的策略user-based policy和基于资源的策略resource-based policy。user-based policy是与身份user/role相关的policy只对真实存在的身份有效。而resource-based policy与具体资源相关包括bucket acl、object acl及bucket policy。两者结合可以对usergroup包括匿名用户来实现权限控制。 当OSS收到请求可以分为bucket操作或object操作时首先验证请求者是否拥有必要的权限包括所有user-based policy和resource-based policy进行评估以决定是否对该请求进行授权。为了确定请求者是否拥有执行特定操作的权限OSS会在收到请求时按顺序执行以下操作 在运行时对所有相关访问策略 (身份策略、bucket策略、acl) 按照特定上下文进行评估。 身份认证 OSS对请求者身份进行判定如果是阿里云账户发出的请求则跳过这步。如果是其他账户包括RAM、STS、匿名会对这些身份拥有的策略子集进行评估。 资源权限认证 如果请求是针对bucket操作发出的则请求者必须拥有来自bucket拥有者的权限且不能被bucket policy显式拒绝。如果请求是针对object发出的在没有被bucket policy显式拒绝的情况下即使被bucket acl拒绝OSS会额外检查object acl以判断是否对请求授权。 bucket请求 object请求 应用场景 跨账号授权OSS资源 假设企业A账号ID 11223344需要将 OSS 操作权限授予企业B账号ID 12345678的员工Anne。我们使用bucket acl和ram policy 完成授权。 > I. 设置bucket acl为private只允许有bucket owner权限的用户访问。 > II. 企业A创建一个角色角色策略中principal设置为企业B的账户ID以便企业B能够担任该角色。 > III. 企业A将合适的访问授权策略赋予该角色。 > IV. 企业B创建RAM用户Anne并授权其担任该角色。 禁止匿名用户访问 使用PutBucketACL接口设置bucket acl为authenticated-read在此场景下只有正确签名的请求才能访问object实现了除匿名访问外的公共读。 参考链接 授权策略语法OSS访问控制BucketPolicy授权
-
发表了文章 2018-08-29
使用Flume将日志导入OSS
前言 Apache Flume是一个高可用、高可靠的分布式日志采集、聚合与传输的系统。它基于流式的数据传输，架构简单、灵活。它简单可扩展的模型，也适合在线的数据分析。 上图是它的简单数据流模型。Flume的数据流由Event贯穿始终，这个Event由外部的Source生成(如Web Server)，携带日志数据并带有一些额外的信息。Source捕获到Event后会进行格式化，然后会推送到一个或者多个Channel中。可以认为Channel是一个Event的缓冲区，它将负责保存Event直到Sink处理完该事件。 Flume的核心运行组件是Agent，一个Agent是一个JVM，是一个完整的数据收集工具。Flume的Agent由三部分组成：Source、Channel和Sink。其中，Flume目前支持了很多种类型的Sink: HDFS Sink、HBase Sink、Thrift Sink、Avro Sink、Elastic Search Sink等。本文主要介绍如何使用HDFS Sink使Flume能够将数据导入到OSS。 配置 配置主要分三块：Java、Hadoop、Flume。配置Java比较简单，设置好JAVA_HOME即可。 Hadoop 首先需要下载Hadoop(以3.0.1版本为例)。Hadoop从2.9.1与3.0.0开始支持OSS，作为默认支持的文件系统。Hadoop的配置可以参考这篇文章。主要是修改$HADOOP_HOME/ etc/hadoop/core-site.xml配置文件，并且配好fs.oss.endpoint、fs.oss.accessKeyId、fs.oss.accessKeySecret和fs.oss.impl。 Flume 下载Apache Flume 1.8.0，并且修改conf目录下的文件 mv conf/flume-conf.properties.template cpnf/flume-conf.properties mv conf/flume-env.sh.template conf/flume-env.sh 修改conf/flume-env.sh文件，添加JAVA_HOME与FLUME_CLASSPATH，其中，需要将HADOOP的lib与conf目录放入到FLUME_CLASSPATH里面。示例如下 export JAVA_HOME=/usr/lib/jdk1.8.0_152 export FLUME_CLASSPATH=$HADOOP_HOME/etc/hadoop:$HADOOP_HOME/share/hadoop/common:$HADOOP_HOME/share/hadoop/hdfs:$HADOOP_HOME/share/hadoop/tools/lib/* 新增conf/hdfs.properties，在这个里面配置Source、Channel与Sink。假设我们想监控某个目录下新产生的文件，并把新产生的文件写到OSS里面去，我们可以这样配置： LogAgent.sources = apache LogAgent.channels = fileChannel LogAgent.sinks = HDFS #sources config LogAgent.sources.apache.type = spooldir LogAgent.sources.apache.spoolDir = /tmp/logs LogAgent.sources.apache.channels = fileChannel LogAgent.sources.apache.basenameHeader = true LogAgent.sources.apache.basenameHeaderKey = fileName #sinks config LogAgent.sinks.HDFS.channel = fileChannel LogAgent.sinks.HDFS.type = hdfs LogAgent.sinks.HDFS.hdfs.path = oss://{Your Bucket Name}/logs/%Y%m%d LogAgent.sinks.HDFS.hdfs.fileType = DataStream LogAgent.sinks.HDFS.hdfs.writeFormat = TEXT LogAgent.sinks.HDFS.hdfs.filePrefix = %{fileName}.%H:%M:%S LogAgent.sinks.HDFS.hdfs.fileSuffix = .log LogAgent.sinks.HDFS.hdfs.batchSize = 1000 LogAgent.sinks.HDFS.hdfs.rollSize = 0 LogAgent.sinks.HDFS.hdfs.rollCount = 0 LogAgent.sinks.HDFS.hdfs.rollInterval = 30 LogAgent.sinks.HDFS.hdfs.useLocalTimeStamp = true #channels config LogAgent.channels.fileChannel.type = memory LogAgent.channels.fileChannel.capacity = 1000000 LogAgent.channels.fileChannel.transactionCapacity = 10000 注意修改 LogAgent.sinks.HDFS.hdfs.path配置项为用户自己在OSS上路径。 启动Agent bin/flume-ng agent --conf-file conf/hdfs.properties -c conf/ --name LogAgent -Dflume.root.logger=console 这个时候会发现抛出异常： Exception in thread "SinkRunner-PollingRunner-DefaultSinkProcessor" java.lang.NoSuchFieldError: INSTANCE at org.apache.http.conn.ssl.SSLConnectionSocketFactory.<clinit>(SSLConnectionSocketFactory.java:144) at com.aliyun.oss.common.comm.DefaultServiceClient.createHttpClientConnectionManager(DefaultServiceClient.java:244) at com.aliyun.oss.common.comm.DefaultServiceClient.<init>(DefaultServiceClient.java:85) at com.aliyun.oss.OSSClient.<init>(OSSClient.java:209) at org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystemStore.initialize(AliyunOSSFileSystemStore.java:150) at org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem.initialize(AliyunOSSFileSystem.java:318) at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3288) at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:123) at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3337) at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3305) at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:476) at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:260) at org.apache.flume.sink.hdfs.BucketWriter$1.call(BucketWriter.java:252) at org.apache.flume.sink.hdfs.BucketWriter$9$1.run(BucketWriter.java:701) at org.apache.flume.auth.SimpleAuthenticator.execute(SimpleAuthenticator.java:50) at org.apache.flume.sink.hdfs.BucketWriter$9.call(BucketWriter.java:698) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) 这个异常是由于Apache Flume使用的http client版本比较低的缘故，把flume lib目录下的httpclient-4.2.1.jar和httpcore-4.4.1.jar删除，重新启动即可。 然后我们往刚才配置的LogAgent.sources.apache.spoolDir里面放文件，使用如下命令 for((i=0;i<=5;i++)); do echo "Hello, world" >> /tmp/logs/sample.data.$i; sleep 30s; done Flume会每隔30秒钟检查这个目录，最终的效果如下: 参考资料 http://flume.apache.org/FlumeUserGuide.html https://yq.aliyun.com/articles/292792?spm=a2c4e.11155435.0.0.7ccba82fbDwfhK
2020年12月
-
12.21 16:37:32
发表了文章
2020-12-21 16:37:32
一站式极致读吞吐，加速业务数字化转型 - 阿里云对象存储 OSS加速器解析与实践
一站式极致读吞吐，加速业务数字化转型 - 阿里云对象存储 OSS加速器解析与实践
2019年07月
-
07.15 14:44:31发表了文章
2019-07-15 14:44:31
如何将Elasticsearch的快照备份至OSS
如何将Elasticsearch的快照备份至OSS
-
发表了文章
2020-12-21
一站式极致读吞吐，加速业务数字化转型 - 阿里云对象存储 OSS加速器解析与实践
-
发表了文章
2019-07-15
如何将Elasticsearch的快照备份至OSS
-
发表了文章
2019-04-23
使用Fluentd读写OSS
-
发表了文章
2019-03-01
Apache Hadoop 2.7如何支持读写OSS
-
发表了文章
2019-01-04
使用Logstash将日志导入OSS
-
发表了文章
2018-11-01
Spark读写OSS并使用OSS Select来加速查询
-
发表了文章
2018-10-30
使用Apache Impala(CDH6)查询OSS的数据
-
发表了文章
2018-10-29
CDH5 Hadoop如何支持读写OSS
-
发表了文章
2018-10-29
HDP2.6 Hadoop如何支持读写OSS
-
发表了文章
2018-09-18
在控制台使用OSS Select
-
发表了文章
2018-09-06
OSS访问控制介绍
-
发表了文章
2018-08-29
使用Flume将日志导入OSS
滑动查看更多
暂无更多信息
暂无更多信息