如何在MaxCompute上处理存储在OSS上的开源格式数据

2024-02-20 23:40
文章标签 数据 处理 存储 格式 开源 oss maxcompute

本文主要是介绍如何在MaxCompute上处理存储在OSS上的开源格式数据，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

为什么80%的码农都做不了架构师？>>>   hot3.png

前言

MaxCompute作为使用最广泛的大数据平台，内部存储的数据以EB量级计算。巨大的数据存储量以及大规模计算下高性能数据读写的需求，对于MaxCompute提出了各种高要求及挑战。处在大数据时代，数据的来源多种多样，开源社区经过十几年的发展，百花齐放，各种各样的数据格式不断的出现。 我们的用户也在各个场景上，通过各种计算框架，积累了各种不同格式的数据。怎样将MaxCompute强大的计算能力开放给这些使用开源格式存储沉淀下来的数据，在MaxCompute上挖掘这些数据中的信息，是MaxCompute团队希望解决的问题。

MaxCompute 2.0最近推出的非结构化计算框架【公测阶段】，旨在从存储介质和存储格式两个维度，打通计算与存储的通道。 在之前的文章中，我们已经介绍过怎样在MaxCompute上对存储在OSS上的文本，音频，图像等格式的数据，以及TableStore(OTS)的KV数据进行计算处理。在这里，则将介绍对于各种流行的开源数据格式(ORC, PARQUET, SEQUENCEFILE, RCFILE, AVRO, TEXTFILE等等)，怎样将其存储在OSS上面，并通过非结构化框架在MaxCompute进行处理。

本着不重造轮子的原则，对于绝大部分这些开源数据格式的解析工作，在非结构化框架中会直接调用开源社区的实现，并且无缝的与MaxCompute系统做对接。

1. 创建EXTERNAL TABLE来绑定OSS外部数据

MaxCompute非结构化数据框架通过EXTERNAL TABLE的概念来提供MaxCompute与各种数据的联通，与读取OSS数据的使用方法类似，对OSS数据进行写操作，首先要通过CREATE EXTERNAL TABLE语句创建出一个外部表，而在读取开源数据格式时，创建外表的DDL语句格式如下：

DROP TABLE [IF EXISTS] <external_table>;CREATE EXTERNAL TABLE [IF NOT EXISTS] <external_table>
(<column schemas>)
[PARTITIONED BY (partition column schemas)]
[ROW FORMAT SERDE '<serde class>']
STORED AS <file format>
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@${endpoint}/${bucket}/${userPath}/'

可以看到，这个语法与HIVE的语法是相当接近的，而在这个CREATE EXTERNAL TABLE的ddl语句中，有如下几点要说明：

  1. 首先要特别说明的是这里使用的是STORED AS的关键字，而不是普通非结构化外表用的STORED BY关键字，这也是目前在读取开源兼容数据时独有的。
  2. 外部表的<column schemas> 必须与具体OSS上存储存储数据的schema相符合。
  3. ROW FORMAT SERDE 并非必选选项，只有在使用一些特殊的格式上，比如TEXTFILE时才需要使用。
  4. STORED AS后面接的是文件格式名字， 比如 ORC/PARQUET/RCFILE/SEQUENCEFILE/TEXTFILE 等等。
  5. 最后还要提到的是，在上面这个例子中，我们在LOCATION上使用了OSS明文AK，这只适用于在用户对于AK的保密性不敏感情况下使用。 对于数据安全比较敏感的场景，比如在多用户场景或者弹外集群上，则推荐使用通过STS/RAM体系事先进行鉴权，从而避免使用明文AK。

1.1 范例1： 关联OSS上存储的PARQUET数据

现在再来看一个具体的例子，假设我们有一些PARQUET文件存放在一个OSS路径上，每个文件都是PARQUET格式，存放着schema为16列(4列BINGINT, 4列DOUBLE, 8列STRING)的数据，那么可以通过如下DDL语句来描述：

CREATE EXTERNAL TABLE tpch_lineitem_parquet
(l_orderkey bigint,l_partkey bigint,l_suppkey bigint,l_linenumber bigint,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
)
STORED AS PARQUET
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/parquet_data/';

1.2 范例2：分区表关联OSS上存储的TEXTFILE数据

同样的数据，如果是每行以JSON格式，存储成OSS上TEXTFILE文件；同时，数据在OSS通过多个目录组织，这时是可以使用MaxCompute分区表和数据关联，则可以通过如下DDL语句来描述：

CREATE EXTERNAL TABLE tpch_lineitem_textfile
(l_orderkey bigint,l_partkey bigint,l_suppkey bigint,l_linenumber bigint,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
)
PARTITIONED BY (ds string)
ROW FORMAT serde 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS TEXTFILE
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/';

如果OSS表目录下面的子目录是以Partition Name方式组织，比如:

oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/ds=20170102/'
oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data/ds=20170103/'
...

则可以使用以下DDL语句ADD PARTITION：

ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170102");
ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170103");

如果OSS分区目录不是按这种方式组织，或者根本不在表目录下，比如：

oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170102/;
oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170103/;
...

则可以使用以下DDL语句ADD PARTITION：

ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170102")
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170102/';
ALTER TABLE tpch_lineitem_textfile ADD PARTITION(ds="20170103")
LOCATION 'oss://${accessKeyId}:${accessKeySecret}@oss-cn-hangzhou-zmf.aliyuncs.com/bucket/text_data_20170103/';
...

2. 读取以及处理 OSS 上面的开源格式数据

对比上面的两个范例，可以看出对于不同文件类型，只要简单修改STORED AS后的格式名。在接下来的例子中，我们将只集中描述对上面PARQUET数据对应的外表（tpch_lineitem_parquet）的处理，如果要处理不同的文件类型，只要在DDL创建外表时指定是PARQUET/ORC/TEXTFILE/RCFILE/TEXTFILE即可，处理数据的语句则是一样的。

2.1 直接读取以及处理OSS上面的开源数据

在创建数据外表后，直接对外表就可以进行与普通MaxCompute表的操作，直接对存储在OSS上的数据进行处理，比如：

SELECT l_returnflag,l_linestatus,SUM(l_extendedprice*(1-l_discount)) AS sum_disc_price,AVG(l_quantity) AS avg_qty,COUNT(*) AS count_order
FROM tpch_lineitem_parquet
WHERE l_shipdate <= '1998-09-02'
GROUP BYl_returnflag,l_linestatus;

可以看到，在这里tpch_lineitem_parquet这个外表被当作一个普通的内部表一样使用。唯一不同的只是在MaxCompute内部计算引擎将从OSS上去读取对应的PARQUET数据来进行处理。

但是我们应该强调的是，在这里直接使用外表，每次读取的时候都需要涉及外部OSS的IO操作，并且MaxCompute系统本身针对内部存储做的许多高性能优化都用不上了，所以性能上会有所损失。 所以如果是需要对数据进行反复计算以及对计算的高效性比较敏感的场景上，我们推荐下面这种用法：先将数据导入MaxCompute内部，再进行计算。

注意，上面例子中的tpch_lineitem_textfile表，因为使用了ROW FORMAT + STORED AS，需要手动设置flag（只使用STORED AS，odps.sql.hive.compatible默认为TRUE），再进行读取，否则会有报错。

SELECT * FROM tpch_lineitem_textfile LIMIT 1;
FAILED: ODPS-0123131:User defined function exception - Traceback:
com.aliyun.odps.udf.UDFException: java.lang.ClassNotFoundException: com.aliyun.odps.hive.wrapper.HiveStorageHandlerWrapper--需要手动设置hive兼容flag
set odps.sql.hive.compatible=true;
SELECT * FROM tpch_lineitem_textfile LIMIT 1;
+------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+
| l_orderkey | l_partkey  | l_suppkey  | l_linenumber | l_quantity | l_extendedprice | l_discount | l_tax      | l_returnflag | l_linestatus | l_shipdate | l_commitdate | l_receiptdate | l_shipinstruct | l_shipmode | l_comment |
+------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+
| 5640000001 | 174458698  | 9458733    | 1            | 14.0       | 23071.58        | 0.08       | 0.06       | N            | O            | 1998-01-26 | 1997-11-16   | 1998-02-18    | TAKE BACK RETURN | SHIP       | cuses nag silently. quick |
+------------+------------+------------+--------------+------------+-----------------+------------+------------+--------------+--------------+------------+--------------+---------------+----------------+------------+-----------+

2.2 将OSS上的开源数据导入MaxCompute，再进行计算

  • 首先创建一个与外部表schema一样的内部表tpch_lineitem_internal，然后将OSS上的开源数据导入MaxCompute内部表，以cFile格式存储在MaxCompute内部：
CREATE TABLE tpch_lineitem_internal LIKE tpch_lineitem_parquet;INSERT OVERWRITE TABLE tpch_lineitem_internal
SELECT * FROM tpch_lineitem_parquet;
  • 直接就可以对内部表进行同样的操作：
SELECT l_returnflag,l_linestatus,SUM(l_extendedprice*(1-l_discount)) AS sum_disc_price,AVG(l_quantity) AS avg_qty,COUNT(*) AS count_order
FROM tpch_lineitem_internal
WHERE l_shipdate <= '1998-09-02'
GROUP BYl_returnflag,l_linestatus;

通过这样子将数据先导入系统的情况下，对同样数据的计算就会更高效得多。

4. 结语

开源的种种数据格式往往由各种数据处理生态产生，而MaxCompute非结构化数据处理框架通过实现计算与存储的互联，希望打通阿里云核心计算平台与各种数据的通路。在这个基础上，各种各样依赖于不同数据格式的应用，将能在MaxCompute计算平台上实现，后继我们会对一些具体的这种应用，比如基因计算等，再做一些具体的case study以及介绍。我们也欢迎有对开源数据进行处理分析的更多应用，能在MaxCompute强大计算能力的基础上开花结果。

原文链接

本文为云栖社区原创内容，未经允许不得转载。

转载于:https://my.oschina.net/yunqi/blog/1821933

这篇关于如何在MaxCompute上处理存储在OSS上的开源格式数据的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/729905。

相关文章

大模型研发全揭秘：客服工单数据标注的完整攻略

大模型研发全揭秘：客服工单数据标注的完整攻略

在人工智能（AI）领域，数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者，掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中，工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注，不仅能够帮助提升客服自动化系统的智能化水平，还能优化客户服务流程，提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行
阅读更多...
基于MySQL Binlog的Elasticsearch数据同步实践

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展，我们的业务数据越来越多，单纯使用 MySQL 已经不能满足我们的数据查询需求，例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后，随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步，我们目前的解决方案是建立数据中间表。把需要检索的业务数据，统一放到一张M
阅读更多...
关于数据埋点，你需要了解这些基本知识

关于数据埋点，你需要了解这些基本知识

产品汪每天都在和数据打交道，你知道数据来自哪里吗？ 移动app端内的用户行为数据大多来自埋点，了解一些埋点知识，能和数据分析师、技术侃大山，参与到前期的数据采集，更重要是让最终的埋点数据能为我所用，否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式，可以区分为： 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理：自动程度高的，能解决通用统计，便于统一化管理，但个性化定
阅读更多...
无人叉车3d激光slam多房间建图定位异常处理方案-墙体画线地图切分方案

无人叉车3d激光slam多房间建图定位异常处理方案-墙体画线地图切分方案

墙体画线地图切分方案 针对问题：墙体两侧特征混淆误匹配，导致建图和定位偏差，表现为过门跳变、外月台走歪等 ·解决思路：预期的根治方案IGICP需要较长时间完成上线，先使用切分地图的工程化方案，即墙体两侧切分为不同地图，在某一侧只使用该侧地图进行定位 方案思路 切分原理：切分地图基于关键帧位置，而非点云。 理论基础：光照是直线的，一帧点云必定只能照射到墙的一侧，无法同时照到两侧实践考虑：关
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
异构存储（冷热数据分离）

异构存储（冷热数据分离）

异构存储主要解决不同的数据，存储在不同类型的硬盘中，达到最佳性能的问题。 异构存储Shell操作 （1）查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies （2）为指定路径（数据存储目录）设置指定的存储策略 hdfs storagepolicies -setStoragePo
阅读更多...
HDFS—存储优化（纠删码）

HDFS—存储优化（纠删码）

纠删码原理 HDFS 默认情况下，一个文件有3个副本，这样提高了数据的可靠性，但也带来了2倍的冗余开销。 Hadoop3.x 引入了纠删码，采用计算的方式，可以节省约50％左右的存储空间。 此种方式节约了空间，但是会增加 cpu 的计算。 纠删码策略是给具体一个路径设置。所有往此路径下存储的文件，都会执行此策略。 默认只开启对 RS-6-3-1024k
阅读更多...
Hadoop集群数据均衡之磁盘间数据均衡

Hadoop集群数据均衡之磁盘间数据均衡

生产环境，由于硬盘空间不足，往往需要增加一块硬盘。刚加载的硬盘没有数据时，可以执行磁盘数据均衡命令。（Hadoop3.x新特性） plan后面带的节点的名字必须是已经存在的，并且是需要均衡的节点。 如果节点不存在，会报如下错误： 如果节点只有一个硬盘的话，不会创建均衡计划： （1）生成均衡计划 hdfs diskbalancer -plan hadoop102 （2）执行均衡计划 hd
阅读更多...
阿里开源语音识别SenseVoiceWindows环境部署

阿里开源语音识别SenseVoiceWindows环境部署

SenseVoice介绍 SenseVoice 专注于高精度多语言语音识别、情感辨识和音频事件检测多语言识别： 采用超过 40 万小时数据训练，支持超过 50 种语言，识别效果上优于 Whisper 模型。富文本识别：具备优秀的情感识别，能够在测试数据上达到和超过目前最佳情感识别模型的效果。支持声音事件检测能力，支持音乐、掌声、笑声、哭声、咳嗽、喷嚏等多种常见人机交互事件进行检测。高效推
阅读更多...
【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯，先赞后看哦~🎈🎈 🏆 作者简介：景天科技苑 🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。 🏆《博客》：Python全栈，前后端开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2