forked from fluent/fluent-bit
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Apache Arrow is an efficient columnar data format that is suitable
for statistical analysis, and popular in machine learning community.
https://arrow.apache.org/
With this patch merged, users now can specify 'arrow' as the
compression type like this:
[OUTPUT]
Name s3
Bucket some-bucket
total_file_size 1M
use_put_object On
Compression arrow
which makes Fluent Bit convert the request buffer into Apache Arrow
format before uploading.
Signed-off-by: Fujimoto Seiji <[email protected]>
Reviewed-by: Sutou Kouhei <[email protected]>
Reviewed-by: Wesley Pettit <[email protected]>- Loading branch information
Showing
7 changed files
with
224 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| set(src | ||
| compress.c) | ||
|
|
||
| add_library(out-s3-arrow STATIC ${src}) | ||
|
|
||
| target_include_directories(out-s3-arrow PRIVATE ${ARROW_GLIB_INCLUDE_DIRS}) | ||
| target_link_libraries(out-s3-arrow ${ARROW_GLIB_LDFLAGS}) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,147 @@ | ||
| /* | ||
| * This converts S3 plugin's request buffer into Apache Arrow format. | ||
| * | ||
| * We use GLib binding to call Arrow functions (which is implemented | ||
| * in C++) from Fluent Bit. | ||
| * | ||
| * https://github.com/apache/arrow/tree/master/c_glib | ||
| */ | ||
|
|
||
| #include <arrow-glib/arrow-glib.h> | ||
| #include <inttypes.h> | ||
|
|
||
| /* | ||
| * GArrowTable is the central structure that represents "table" (a.k.a. | ||
| * data frame). | ||
| */ | ||
| static GArrowTable* parse_json(uint8_t *json, int size) | ||
| { | ||
| GArrowJSONReader *reader; | ||
| GArrowBuffer *buffer; | ||
| GArrowBufferInputStream *input; | ||
| GArrowJSONReadOptions *options; | ||
| GArrowTable *table; | ||
| GError *error = NULL; | ||
|
|
||
| buffer = garrow_buffer_new(json, size); | ||
| if (buffer == NULL) { | ||
| return NULL; | ||
| } | ||
|
|
||
| input = garrow_buffer_input_stream_new(buffer); | ||
| if (input == NULL) { | ||
| g_object_unref(buffer); | ||
| return NULL; | ||
| } | ||
|
|
||
| options = garrow_json_read_options_new(); | ||
| if (options == NULL) { | ||
| g_object_unref(buffer); | ||
| g_object_unref(input); | ||
| return NULL; | ||
| } | ||
|
|
||
| reader = garrow_json_reader_new(GARROW_INPUT_STREAM(input), options, &error); | ||
| if (reader == NULL) { | ||
| g_error_free(error); | ||
| g_object_unref(buffer); | ||
| g_object_unref(input); | ||
| g_object_unref(options); | ||
| return NULL; | ||
| } | ||
|
|
||
| table = garrow_json_reader_read(reader, &error); | ||
| if (table == NULL) { | ||
| g_error_free(error); | ||
| g_object_unref(buffer); | ||
| g_object_unref(input); | ||
| g_object_unref(options); | ||
| g_object_unref(reader); | ||
| return NULL; | ||
| } | ||
| g_object_unref(buffer); | ||
| g_object_unref(input); | ||
| g_object_unref(options); | ||
| g_object_unref(reader); | ||
| return table; | ||
| } | ||
|
|
||
| static GArrowResizableBuffer* table_to_buffer(GArrowTable *table) | ||
| { | ||
| GArrowResizableBuffer *buffer; | ||
| GArrowBufferOutputStream *sink; | ||
| GError *error = NULL; | ||
| gboolean success; | ||
|
|
||
| buffer = garrow_resizable_buffer_new(0, &error); | ||
| if (buffer == NULL) { | ||
| g_error_free(error); | ||
| return NULL; | ||
| } | ||
|
|
||
| sink = garrow_buffer_output_stream_new(buffer); | ||
| if (sink == NULL) { | ||
| g_object_unref(buffer); | ||
| return NULL; | ||
| } | ||
|
|
||
| success = garrow_table_write_as_feather( | ||
| table, GARROW_OUTPUT_STREAM(sink), | ||
| NULL, &error); | ||
| if (!success) { | ||
| g_error_free(error); | ||
| g_object_unref(buffer); | ||
| g_object_unref(sink); | ||
| return NULL; | ||
| } | ||
| g_object_unref(sink); | ||
| return buffer; | ||
| } | ||
|
|
||
| int out_s3_compress_arrow(uint8_t *json, size_t size, void **out_buf, size_t *out_size) | ||
| { | ||
| GArrowTable *table; | ||
| GArrowResizableBuffer *buffer; | ||
| GBytes *bytes; | ||
| gconstpointer ptr; | ||
| gsize len; | ||
| uint8_t *buf; | ||
|
|
||
| table = parse_json(json, size); | ||
| if (table == NULL) { | ||
| return -1; | ||
| } | ||
|
|
||
| buffer = table_to_buffer(table); | ||
| g_object_unref(table); | ||
| if (buffer == NULL) { | ||
| return -1; | ||
| } | ||
|
|
||
| bytes = garrow_buffer_get_data(GARROW_BUFFER(buffer)); | ||
| if (bytes == NULL) { | ||
| g_object_unref(buffer); | ||
| return -1; | ||
| } | ||
|
|
||
| ptr = g_bytes_get_data(bytes, &len); | ||
| if (ptr == NULL) { | ||
| g_object_unref(buffer); | ||
| g_bytes_unref(bytes); | ||
| return -1; | ||
| } | ||
|
|
||
| buf = malloc(len); | ||
| if (buf == NULL) { | ||
| g_object_unref(buffer); | ||
| g_bytes_unref(bytes); | ||
| return -1; | ||
| } | ||
| memcpy(buf, ptr, len); | ||
| *out_buf = (void *) buf; | ||
| *out_size = len; | ||
|
|
||
| g_object_unref(buffer); | ||
| g_bytes_unref(bytes); | ||
| return 0; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| /* | ||
| * This function converts out_s3 buffer into Apache Arrow format. | ||
| * | ||
| * `json` is a string that contain (concatenated) JSON objects. | ||
| * | ||
| * `size` is the length of the json data (excluding the trailing | ||
| * null-terminator character). | ||
| * | ||
| * Return 0 on success (with `out_buf` and `out_size` updated), | ||
| * and -1 on failure | ||
| */ | ||
|
|
||
| int out_s3_compress_arrow(char *json, size_t size, void **out_buf, size_t *out_size); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters