Shulou(Shulou.com)05/31 Report--

Today, I will talk to you about how to use Function Compute to simply clean the data in the table storage. Many people may not know much about it. In order to make you understand better, the editor has summarized the following contents for you. I hope you can get something according to this article.

Function Computing (Function Compute) is an event-driven service. Through function computing, users do not need to manage the operation of the server, but only need to write code and upload it. The function calculation prepares the computing resources and runs the user code in an elastic way, while the user only needs to pay according to the resources consumed by the actual code.

Table Store Stream is a data channel used to obtain incremental data in Table Store table. By creating Table Store trigger, we can realize the automatic docking of Table Store Stream and function calculation, and let the self-defined program logic in calculation function automatically deal with the data modification in Table Store table.

The high concurrent writing performance and low storage cost of table storage are very suitable for the storage of Internet of things, logs and monitoring data. We can write the data into the table storage. At the same time, in the function calculation, we do simple cleaning, conversion, aggregation calculation and other operations on the new data, and write the cleaned data back to the result table stored in the table, and provide real-time access to the original detail data and result data.

Next, we use function calculations to simply clean the data stored in the table and write it to the result table.

Data definition

We assume that what is written is log data, including three basic fields:

The field name type means the level of id integer log idlevel integer log. The larger the level is, the higher the level is. The content of message string log

We need to write the log of level > 1 to another data table to use as a special query.

Implementation process: create instances and data tables

Create a table storage instance in the console of the table storage (_ _ take East China 2 distribute-test as an example _ _), and create the source table (_ _ source_data__) and the result table (_ _ result__). The primary key is uniform _ _ id (integer) _ _. Since the table storage is a schemafree structure, there is no need to define other attribute column fields in advance.

Enable the Stream function of the data source table

The trigger function needs to turn on the Stream function of the data table before it can process the incremental data written into the table storage in the function calculation.

The Stream record expires for the maximum amount of incremental data that can be read through the StreamAPI.

Since triggers can only bind existing functions, create services and functions on the same region on the console where the functions are calculated first.

Create a function calculation service

To create service and processing functions on the console of function calculation, we continue to use East China 2 nodes.

1. Create a service at 2 nodes in East China.

two。 Select the creation function in turn: blank function-- do not create a trigger.

The function name is: etl_test, select the python2.7 environment, and edit the code online

The function entry is: etl_test.handler

Edit the code later and click next.

3. Conduct service authorization

Since the function calculation needs to write the running log to the log service, and the table stored in the table needs to be read and written, the function calculation needs to be authorized. For convenience, we first add AliyunOTSFullAccess and _ _ AliyunLogFullAccess _ _ permissions. In actual production, it is recommended to add permissions according to the principle of minimum permissions.

4. Click Authorization complete and create the function.

5. Modify the function code.

After creating the function, click the corresponding function-Code execution, edit the code and save it, where INSTANCE_NAME (instance name stored in the table) and REGION (region used) need to be modified according to the situation:

The sample code to use is as follows:

#! / usr/bin/env python#-*-coding: utf-8-*-import cborimport jsonimport tablestore as otsINSTANCE_NAME = 'distribute-test'REGION =' cn-shanghai'ENDPOINT = 'http://%s.%s.ots-internal.aliyuncs.com'%(INSTANCE_NAME, REGION) RESULT_TABLENAME =' result'def _ utf8 (input): return str (bytearray (input, "utf-8")) def get_attrbute_value (record Column): attrs = record [u 'Columns'] for x in attrs: if x [u' ColumnName'] = = column: return x ['Value'] def get_pk_value (record, column): attrs = record [u' PrimaryKey'] for x in attrs: if x ['ColumnName'] = column: return x [' Value'] # because AliyunOTSFullAccess permission has been granted The credentials obtained here has access to table storage def get_ots_client (context): creds = context.credentials client = ots.OTSClient (ENDPOINT, creds.accessKeyId, creds.accessKeySecret, INSTANCE_NAME, sts_token = creds.securityToken) return clientdef save_to_ots (client, record): id = int (get_pk_value (record, 'id')) level = int (get_attrbute_value (record,' level')) msg = get_attrbute_value (record) 'message') Competition = [(_ utf8 (' id'), id),] attr = [(_ utf8 ('level'), level), (_ utf8 (' message'), _ utf8 (msg)),] row = ots.Row (Competition, attr) client.put_row (RESULT_TABLENAME, row) def handler (event) Context): records = cbor.loads (event) # records = json.loads (event) client = get_ots_client (context) for record in records ['Records']: level = int (get_attrbute_value (record,' level')) if level > 1: save_to_ots (client, record) else: print "Level

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.