Integrate CloudFlow with OTS to implement read/write and persistent storage of data in workflows-CloudFlow(CloudFlow)-阿里云帮助中心

You can integrate CloudFlow with Tablestore to permanently store data in the running of workflows to tables of Tablestore or use the data in tables of Tablestore as the data source of workflows. This topic describes the sample code, parameters, and output of data reading and writing when CloudFlow is integrated with Tablestore.

Background information

Tablestore is a cost-effective table-based serverless storage service that can be used to store large volumes of structured data. Tablestore allows you to query and retrieve online data in milliseconds and perform multi-dimensional analysis on stored data. Tablestore is suitable for scenarios such as billing, instant messaging (IM), IoT, Internet of Vehicles (IoV), risk control, and intelligent recommendation. Tablestore provides a deeply optimized all-in-one storage solution for IoT applications. For more information, see What is a Tablestore?

Execution mode

When you integrate CloudFlow with Tablestore, only the RequestComplete service integration mode can be used.

Read data from Tablestore

GetRow

Reads data in a single row based on a specific primary key.

Integration example

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: GetRow
States:
  - Type: Task
    Name: GetRow
    Action: Tablestore:GetRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey.$: $Input.primaryKey
      columnsToGet:
        - COL1
    End: true

Parameter description

Parameter	Required	Description	Example
region	No	The region in which the Tablestore instance resides. The default value is the region in which the current workflow resides.	cn-hangzhou
instanceName	Yes	The name of the Tablestore instance.	test-instance-name
tableName	Yes	The name of the table in Tablestore.	test_table_name
primaryKey	Yes	The primary key of the data. Tablestore supports multiple primary key columns. Type of this parameter value: object.	`primaryKey: pkCol1: val1 pkCol2: val2`
columnsToGet	No	The names of the columns that you want to query and return. Type of this parameter value: array.	`columnsToGet: - col1 - col2`
startColumn	No	The start column of the read operation in the row. This parameter is used for wide-column reads. The columns are sorted based on their names in alphabetical order. The response contains the start column.	col1
endColumn	No	The end column of the read operation in the row. The response does not contain the end column. The columns are sorted based on their names in alphabetical order.	col10

Output of the integration

{
    "requestId":"00060e6a-7bdb-786f-5d68-******",
    "row":{
        "COL1":"val2",
        "PK1":"aaa",
        "PK2":1699949934946000
    }
}

GetRange

Reads data within a specific value range of the primary key.

Integration example

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: GetRange
States:
  - Type: Task
    Name: GetRange
    Action: Tablestore:GetRange
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      startPrimaryKey:
        - key: pkCol1
          valueType: value
          value.$: $Input.value1
        - key: pkCol2
          valueType: min
      endPrimaryKey:
        - key: pkCol1
          valueType: value
          value.$: $Input.value1
        - key: pkCol2
          valueType: max
      columnsToGet:
       - COL1
    End: true

Parameter description

Parameter	Required	Description	Example
region	No	The region in which the Tablestore instance resides. The default value is the region in which the current workflow resides.	cn-hangzhou
instanceName	Yes	The name of the Tablestore instance.	test-instance-name
tableName	Yes	The name of the table in Tablestore.	test_table_name
startPrimaryKey	Yes	The start primary key of the read operation. Type of this parameter value: object. valueType: the type of the primary key value. valid values: min: minimum value, max: maximum value, value: custom value, auto: auto-incremented value. value: the specific value when `valueType = value` is set.	`startPrimaryKey: - key: pkCol1 valueType: value value: val1 - key: pkCol2 valueType: min`
endPrimaryKey	Yes	The end primary key of the read operation.	`endPrimaryKey: - key: pkCol1 valueType: value value: val1 - key: pkCol2 valueType: max`
columnsToGet	No	The names of the columns that you want to query and return. Type of this parameter value: array.	`columnsToGet: - col1 - col2`
startColumn	No	The start column of the read operation in the row. This parameter is used for wide-column reads. The columns are sorted based on their names in alphabetical order. The response contains the start column.	col1
endColumn	No	The end column of the read operation in the row. The response does not contain the end column. The columns are sorted based on their names in alphabetical order.	col10
limit	No	The maximum number of rows that you want to return. The value of this parameter must be greater than 0.	10

Output of the integration

{
    "requestId":"00060e6a-88c3-ecd1-c1cb-******",
    "rows":[
        {
            "COL1":"val2",
            "PK1":"aaa",
            "PK2":1699881177699000
        },
        {
            "COL1":"val2",
            "PK1":"aaa",
            "PK2":1699881296484000
        }
    ]
}

Write data to CloudFlow

PutRow

Inserts a row into the table. If the row exists, Tablestore deletes all versions of data in all columns from the existing row and writes new data.

Integration example

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: PutRow
States:
  - Type: Task
    Name: PutRow
    Action: Tablestore:PutRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey:
        PK1: aaa
      rowData:
        COL1: val1
      condition: IGNORE
    End: true

Parameter description

Parameter	Required	Description	Example
region	No	The region in which the Tablestore instance resides. The default value is the region in which the current workflow resides.	cn-hangzhou
instanceName	Yes	The name of the Tablestore instance.	test-instance-name
tableName	Yes	The name of the table in Tablestore.	test_table_name
primaryKey	Yes	The primary key of the data. Tablestore supports multiple primary key columns. Type of this parameter value: object.	`primaryKey: pkCol1: val1 pkCol2: val2`
rowData	No	The row data that you want to write to CloudFlow. Do not include the primary keys into this parameter value.	`rowData: col3: val3 col4: val4`
condition	No	Specifies whether to check the existence of the row before data is written. Default value: IGNORE. Valid values: IGNORE: does not check the existence of the row. EXPECT_EXIST: expects that the row exists. EXPECT_NOT_EXIST: does not expect that the row exists.	IGNORE

Output of the integration

{
    "requestId":"00060e6a-8c81-9ac4-5d68-******",
    "row":{
        "COL1":"val2",
        "PK1":"aaa",
        "PK2":1704700646764000
    }
}

UpdateRow

Updates the data of a row. You can insert attribute columns into a row, delete attribute columns from a row, delete data of specific versions of attribute columns, or update the values of existing attribute columns. If the row does not exist, a new row is inserted.

Integration example

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: UpdateRow
States:
  - Type: Task
    Name: UpdateRow
    Action: Tablestore:UpdateRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey.$: $Input.primaryKey
      rowData:
        COL1: val2
      condition: EXPECT_EXIST
    End: true

Parameter description

Parameter	Required	Description	Example
region	No	The region in which the Tablestore instance resides. The default value is the region in which the current workflow resides.	cn-hangzhou
instanceName	Yes	The name of the Tablestore instance.	test-instance-name
tableName	Yes	he name of the table in Tablestore.	test_table_name
primaryKey	Yes	The primary key of the data. Tablestore supports multiple primary key columns. Type of this parameter value: object.	`primaryKey: pkCol1: val1 pkCol2: val2`
rowData	No	The row data that you want to write to CloudFlow. Do not include the primary keys into this parameter value.	`rowData: col3: val3 col4: val4`
condition	No	Specifies whether to check the existence of the row before data is written. Default value: IGNORE. Valid values: IGNORE: does not check the existence of the row. EXPECT_EXIST: expects that the row exists. EXPECT_NOT_EXIST: does not expect that the row exists.	IGNORE

Output of the integration

{
    "requestId":"00060e6a-8c81-9ac4-5d68-******",
    "row":{
        "COL1":"val2",
        "PK1":"aaa",
        "PK2":1704700646764000
    }
}

DeleteRow

Deletes a row of data.

Integration example

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: DeleteRow
States:
  - Type: Task
    Name: DeleteRow
    Action: Tablestore:DeleteRow
    TaskMode: RequestComplete
    Parameters:
      region: cn-hangzhou
      instanceName: test-instance-name
      tableName: test_table_name
      primaryKey.$: $Input.primaryKey
    End: true

Parameter description

Parameter	Required	Description	Example
region	No	The region in which the Tablestore instance resides. The default value is the region in which the current workflow resides.	cn-hangzhou
instanceName	Yes	The name of the Tablestore instance.	test-instance-name
tableName	Yes	The name of the table in Tablestore.	test_table_name
primaryKey	Yes	The primary key of the data. Tablestore supports multiple primary key columns. Type of this parameter value: object.	`primaryKey: pkCol1: val1 pkCol2: val2`
condition	No	Specifies whether to check the existence of the row before data is written. Default value: IGNORE. Valid values: IGNORE: does not check the existence of the row. EXPECT_EXIST: expects that the row exists. EXPECT_NOT_EXIST: does not expect that the row exists.	IGNORE

Output of the integration

{
    "requestId":"00060e6a-939b-965b-1069-******"
}