Python SDK

Create New Data Ingestion

Create data ingestion in the Dashboard or through the command-line tool, using Auto Push as the data source type.

Import Python SDK

The Python SDK is built by encapsulating the C++ SDK and exposes relevant interfaces. At runtime, it calls the underlying C++ SDK to implement related operations (such as sending, closing, etc.). Because the underlying layer runs based on the C++ SDK, it is highly bound to the system's C++ environment. If the C++ environment used by the provided software package is not completely consistent with the user's system environment, it may cause runtime errors. First, you need to compile it from the source code yourself. See C++ SDK Compilation and Usage. If compilation is successful, you will find the static library file dataproxy_sdk.a in the dataproxy-sdk-cpp/release/lib directory. Enter the Python SDK directory dataproxy-sdk-python and execute the following commands in sequence:

chmod +x ./build.sh
./build.sh

When the .so file is generated, you will see the following message, and you can choose to enter the target directory of the .so file. By default, the .so file will be copied to the system python site-packages directory.

Your system's Python site-packages directory is: xxx/xxx
Enter the target directory for the .so files (Press Enter to use the default site-packages directory):

After the build process is completed, you can import the package in your Python project and use the InLong DataProxy SDK to report data.

import inlong_dataproxy

Note: When the C++ SDK or the Python version you are using is updated, you need to rebuild it following the steps above.

Data Reporting Process

After introducing the SDK, you can report single (or batch) data by calling the SDK's send related interfaces. For sending demo, please refer to send_demo.py. The overall process includes the following three steps:

Create SDK Instance

The SDK supports the creation of multiple SDK instances within a process. Each SDK instance is independent, and the SDK instance is thread-safe:

• Create an SDK instance object

inlong_api = inlong_dataproxy.InLongApi()

• Configure GroupID, StreamID task parameters, auth_id, auth_auth_key user CMK authentication parameters, and set is_internal to true in the configuration file. For detailed configuration file instructions, see the appendix.

// Initialize the SDK, with the parameter being the path name of the configuration file; a return value of zero indicates successful initialization.
init_status = inlong_api.init_api("./config.json")

Call the Send Interface for Data Reporting

The SDK supports single (recommended) and batch sending, both of which are asynchronous modes, and the data reporting interface is thread-safe. Before reporting data, you can set a callback function to handle callbacks when data sending fails. The callback function signature is as follows:

int CallbackFunc(const char *a, const char *b, const char *c, int32_t d, const int64_t e, const char *f)

• Single Data Reporting Interface

// Return value: Zero indicates successful sending, non-zero indicates failure. For specific exception return values, please refer to SDKInvalidResult in tc_api.h of the C++ SDK.
// msg is the data to be sent, msg_len is the length of the data, and call_back_func is the callback function (which can be set to None).
send(inlong_group_id, inlong_stream_id, msg, msg_len, call_back_func)

Close SDK

Call the close interface to close the SDK:

// A return value of zero indicates successful closure, and subsequent data reporting will not be possible.
// max_waitms: The maximum number of milliseconds to wait before closing the SDK, waiting for the internal data sending of the SDK to complete.
inlong_api.close_api(max_waitms)

Notes

• It is recommended to use the SDK as a resident service for data reporting, to avoid frequent initialization and closure in the same process midway. Repeated initialization and closure will bring more overhead.
• The SDK sending is carried out asynchronously, and a return value of 0 indicates that the data has been successfully stored in the SDK internal buffer, waiting for network transmission. If the inlong_group_id itself is incorrectly configured or there is a network exception, it will also lead to data sending failure. Therefore, it is recommended that users set a callback when calling this interface, and execute the callback when data fails to be sent after multiple retries.

Appendix: Configuration File Parameter Description

Parameter	Meaning	Default Value
is_internal	Company Internal Use Flag	true
inlong_group_ids	List of inlong_group_id	b_inlong_group_test_01, b_inlong_group_test_02
manager_url	Manager Address	http://127.0.0.1:8099/inlong/manager/openapi/dataproxy/getIpList
manager_update_interval	Request Manager Interval	2 minutes
manager_url_timeout	Pushed Data Time	Timestamp in milliseconds
msg_type	Data Type	7
max_proxy_num	Maximum DataProxy Data per Instance	8
per_groupid_thread_nums	Number of Threads per inlong_group_id	1
dispatch_interval_zip	Compression Data Interval Period	8 ms
dispatch_interval_send	Data Distribution Interval	10 ms
recv_buf_size	Receive Buffer Size	10240000 bytes
send_buf_size	Send Buffer Size	10240000 bytes
enable_pack	Whether to Allow Packing	true
pack_size	Packing Size	409600 bytes
pack_timeout	Packing Interval	3000 ms
ext_pack_size	Maximum Size of Single Message	409600 bytes
enable_zip	Whether to Allow Compression	true
min_zip_len	Minimum Compression Size	512 bytes
tcp_detection_interval	TCP Detection Period	60000 ms
tcp_idle_time	TCP Idle Cycle	600000 ms
log_num	Number of Log Files	10
log_size	Size of Single Log File	104857600 bytes
log_level	Log Level	3. Log Levels: trace(4)>debug(3)>info(2)>warn(1)>error(0)
log_path	Log Directory	./
need_auth	Whether to Enable Authentication	true
auth_id	Account
auth_key	Key