当前位置: 首页 > 工具软件 > Python_sdk > 使用案例 >

python的sdk是什么_Python SDK介绍

袁安志
2023-12-01

Python SDK介绍

安装

快速安装

$ sudo pip install pydatahub

源码安装

$ git clone https://github.com/aliyun/aliyun-datahub-sdk-python.git

$ cd aliyun-datahub-sdk-python

$ sudo python setup.py install

常见问题

1.如果安装过程中出现错误信息’Python.h: No such file or directory’,常用的操作系统安装方式如下:

$ sudo apt-getinstall python-dev# for python2.x installs

$ sudo apt-getinstall python3-dev# for python3.x installs

$ sudo yum install python-devel# for python2.x installs

$ sudo yum install python34-devel# for python3.4 installs

2.如果使用windows操作系统,根据提示信息可到 此处 下载安装对应版本的 Visual C++ SDK。

Windows 10 安装cprotobuf依赖时如果报类似如下错误,也表示需要安装Visual C++ 生成工具:

bulding'cprotobuf.internal'extention

error:[WinError2]Thesystem cannot find the file specified

推荐使用python3.6或以上,会明确提示所需版本及链接信息。

3.Windows 下如果安装依赖时报类似如下错误,是环境问题所致,请搜索相关错误,根据具体情况,拷贝所需文件,或是直接使用 developer command prompt 工具进行安装:

LINK:fatal error LNK1158:cannot run'rc.exe'

4.网络环境受限无法自动使用pip安装依赖时,可以从github中下载相关依赖,并按照readme进行安装。

安装验证

$ python-c"from datahub import DataHub"

如果上述命令执行成功,恭喜你安装Datahub Python版本SDK成功!

基本概念

准备工作

访问DataHub服务需要使用阿里云认证账号,需要提供阿里云accessId及accessKey。 同时需要提供访问的服务地址。

创建Project或使用SDK接口进行创建

初始化Datahub

importsys

importtraceback

fromdatahubimportDataHub

fromdatahub.exceptionsimportResourceExistException

fromdatahub.modelsimportFieldType,RecordSchema,TupleRecord,BlobRecord,CursorType,RecordType

access_id=***your access id***

access_key=***your access key***

endpoint=***your datahub server endpoint***

dh=DataHub(access_id,access_key,endpoint)

Project操作

创建示例

project_name='project'

comment='comment'

try:

dh.create_project(project_name,comment)

print("create project success!")

print("=======================================\n\n")

exceptResourceExistException:

print("project already exist!")

print("=======================================\n\n")

exceptExceptionase:

print(traceback.format_exc())

sys.exit(-1)

Topic操作

Tuple Topic

Tuple类型Topic写入的数据是有格式的,需要指定Record Schema,目前支持以下几种数据类型:

类型

含义

值域

Bigint

8字节有符号整型。请不要使用整型的最小值 (-9223372036854775808),这是系统保留值。

-9223372036854775807 ~ 9223372036854775807

String

字符串,只支持UTF-8编码。

单个String列最长允许1MB。

Boolean

布尔型。

可以表示为True/False,true/false, 0/1

Double

8字节双精度浮点数。

-1.0 10308 ~ 1.0 10308

TimeStamp

时间戳类型

表示到微秒的时间戳类型

创建示例

topic_name="tuple_topic"

shard_count=3

life_cycle=7

record_schema=RecordSchema.from_lists(

['bigint_field','string_field','double_field','bool_field','time_field'],

[FieldType.BIGINT,FieldType.STRING,FieldType.DOUBLE,FieldType.BOOLEAN,FieldType.TIMESTAMP])

try:

dh.create_tuple_topic(project_name,topic_name,shard_count,life_cycle,record_schema,comment)

print("create tuple topic success!")

print("=======================================\n\n")

exceptResourceExistException:

print("topic already exist!")

print("=======================================\n\n")

exceptExceptionase:

print(traceback.format_exc())

sys.exit(-1)

Blob Topic

Blob类型Topic支持写入一块二进制数据作为一个Record,数据将会以BASE64编码传输。

topic_name="blob_topic"

shard_count=3

life_cycle=7

try:

dh.create_blob_topic(project_name,topic_name,shard_count,life_cycle,comment)

print("create blob topic success!")

print("=======================================\n\n")

exceptResourceExistException:

print("topic already exist!")

print("=======================================\n\n")

exceptExceptionase:

print(traceback.format_exc())

sys.exit(-1)

数据发布/订阅

获取Shard列表

list_shards接口获取topic下的所有shard

shard_result=dh.list_shard(project_name,topic_name)

shards=shard_result.shards

print(len(shards))

返回结果是一个ListShardResult对象,包含一个Shard对象的list,list中的每个元素是一个shard,可以获取shard_id,state状态,begin_hash_key,end_hash_key等信息

发布数据

put_records接口向一个topic发布数据

put_result=dh.put_records(project_name,topic_name,records)

print(put_result.failed_record_count)

print(put_result.failed_records)

其中传入参数records是一个List对象,每个元素为一个record,但是必须为相同类型的record,即Tuple类型或者Blob类型,返回结果为PutRecordsResult对象,包含failed_record_count和failed_records成员,failed_records是一个FailedRecord对象的list,FailedRecord对象包含成员index,error_code和error_message

写入Tuple类型Record示例

try:

# block等待所有shard状态ready

dh.wait_shards_ready(project_name,topic_name)

print("shards all ready!!!")

print("=======================================\n\n")

topic_result=dh.get_topic(project_name,topic_name)

print(topic_result)

iftopic_result.record_type!=RecordType.TUPLE:

print("topic type illegal!")

sys.exit(-1)

print("=======================================\n\n")

record_schema=topic_result.record_schema

records0=[]

record0=TupleRecord(schema=record_schema,values=[1,'yc1',10.01,True,1455869335000000])

record0.shard_id='0'

record0.put_attribute('AK','47')

records0.append(record0)

record1=TupleRecord(schema=record_schema)

record1.set_value('bigint_field',2)

record1.set_value('string_field','yc2')

record1.set_value('double_field',None)

record1.set_value('bool_field',False)

record1.set_value('time_field',1455869335000011)

record1.hash_key='4FFFFFFFFFFFFFFD7FFFFFFFFFFFFFFD'

records0.append(record1)

record2=TupleRecord(schema=record_schema)

record2.set_value(0,3)

record2.set_value(1,'yc3')

record2.set_value(2,1.1)

record2.set_value(3,False)

record2.set_value(4,1455869335000011)

record2.attributes={'key':'value'}

record2.partition_key='TestPartitionKey'

records0.append(record2)

put_result=dh.put_records(project_name,topic_name,records0)

print(put_result)

print("put tuple %d records, failed count: %d"%(len(records0),put_result.failed_record_count))

# failed_record_count如果大于0最好对failed record再进行重试

print("=======================================\n\n")

exceptDatahubExceptionase:

print(e)

sys.exit(-1)

写入BLOB类型Record示例

try:

records1=[]

record3=BlobRecord(blob_data='data')

record3.shard_id='0'

record3.put_attribute('a','b')

records1.append(record3)

put_result=dh.put_records(project_name,topic_name,records1)

print(put_result)

exceptDatahubExceptionase:

print(e)

sys.exit(-1)

获取cursor

获取Cursor,可以通过三种方式获取:OLDEST, LATEST, SYSTEM_TIME

OLDEST: 表示获取的cursor指向当前有效数据中时间最久远的record

LATEST: 表示获取的cursor指向当前最新的record

SYSTEM_TIME: 表示获取的cursor指向大于等于该时间(单位毫秒)的第一条record

shard_id='0'

time_stamp=0

cursor_result0=dh.get_cursor(project_name,topic_name,shard_id,CursorType.OLDEST)

cursor_result1=dh.get_cursor(project_name,topic_name,shard_id,CursorType.LATEST)

cursor_result2=dh.get_cursor(project_name,topic_name,shard_id,CursorType.SYSTEM_TIME,time_stamp)

cursor=cursor_result0.cursor

通过get_cursor接口获取用于读取指定位置之后数据的cursor

订阅数据

从指定shard读取数据,需要指定从哪个Cursor开始读,并指定读取的上限数据条数,如果从Cursor到shard结尾少于Limit条数的数据,则返回实际的条数的数据。

project_name='project'

shard_id="0"

limit=10

# 读取blob topic的record

topic_name='blob_topic'

get_result=dh.get_blob_records(project_name,topic_name,shard_id,cursor,limit)

# 读取tuple topic的record

topic_name='tuple_topic'

get_result=dh.get_tuple_records(project_name,topic_name,shard_id,record_schema,cursor,limit)

消费Tuple类型Record示例

try:

# block等待所有shard状态ready

dh.wait_shards_ready(project_name,topic_name)

print("shards all ready!!!")

print("=======================================\n\n")

topic_result=dh.get_topic(project_name,topic_name)

print(topic_result)

iftopic_result.record_type!=RecordType.TUPLE:

print("topic type illegal!")

sys.exit(-1)

print("=======================================\n\n")

shard_id='0'

limit=10

cursor_result=dh.get_cursor(project_name,topic_name,shard_id,CursorType.OLDEST)

cursor=cursor_result.cursor

whileTrue:

get_result=dh.get_tuple_records(project_name,topic_name,shard_id,record_schema,cursor,limit)

forrecordinget_result.records:

print(record)

if0==get_result.record_count:

time.sleep(1)

cursor=get_result.next_cursor

exceptDatahubExceptionase:

print(e)

sys.exit(-1)

结尾

 类似资料: