-
Notifications
You must be signed in to change notification settings - Fork 31
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor Sink Writer to accept blocks inplace of bytes (#86)
* change Write to use Stream API and impl Stream for bq Sink * basic refactoring to support list of sinks and single encoder * bump go version in dockerfile * added Filter for slice of rows * added pubsub compaction implementation with Manoj * Fix test cases * Fix all test cases * Remove print line * Add yaml inline tag to BaseSink (#3) * Add yaml inline tag to BaseSink * Add inline tag to BaseSink * Revert sample yaml configuration * Refactor/encoder (#4) * Add yaml inline tag to BaseSink * Add inline tag to BaseSink * Revert sample yaml configuration * fix conflict * Refactor/encoder (#5) * Add yaml inline tag to BaseSink * Add inline tag to BaseSink * Revert sample yaml configuration * fix conflict * add contributors in readme Co-authored-by: sosoonyuan <[email protected]> Co-authored-by: JeffreyLean <[email protected]> Co-authored-by: Jeffrey lean <[email protected]>
- Loading branch information
4 people
authored
Jun 7, 2022
1 parent
c6fb73c
commit d40b2a4
Showing
41 changed files
with
868 additions
and
320 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -13,5 +13,6 @@ testdata-* | |
| .envrc | ||
| logs/ | ||
| *.venv | ||
| .vscode/ | ||
| talaria | ||
| *.so | ||
| *.so | ||
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,15 @@ | ||
| { | ||
| // Use IntelliSense to learn about possible attributes. | ||
| // Hover to view descriptions of existing attributes. | ||
| // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387 | ||
| "version": "0.2.0", | ||
| "configurations": [ | ||
| { | ||
| "name": "Test current file", | ||
| "type": "go", | ||
| "request": "launch", | ||
| "mode": "test", | ||
| "program": "${file}" | ||
| } | ||
| ] | ||
| } |
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 |
|---|---|---|
| @@ -1,4 +1,4 @@ | ||
| FROM golang:1.16 AS builder | ||
| FROM golang:1.17 AS builder | ||
|
|
||
| ARG GO111MODULE="on" | ||
| ARG GOOS="linux" | ||
|
|
||
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
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,31 @@ | ||
| // Copyright 2019-2020 Grabtaxi Holdings PTE LTE (GRAB), All rights reserved. | ||
| // Use of this source code is governed by an MIT-style license that can be found in the LICENSE file | ||
|
|
||
| package block | ||
|
|
||
| import ( | ||
| "github.com/kelindar/talaria/internal/encoding/typeof" | ||
| ) | ||
|
|
||
| // FromBlockBy creates and returns a list of new block.Row for a block. | ||
| func FromBlockBy(blk Block, schema typeof.Schema) ([]Row, error) { | ||
| cols, err := blk.Select(schema) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| rowCount := cols.Any().Count() | ||
| rows := make([]Row, rowCount) | ||
| for i := 0; i < rowCount; i++ { | ||
| row := NewRow(schema, len(schema)) | ||
| for name := range schema { | ||
| col := cols[name] | ||
| val := col.At(i) | ||
| if val == nil { | ||
| continue | ||
| } | ||
| row.Set(name, val) | ||
| } | ||
| rows[i] = row | ||
| } | ||
| return rows, nil | ||
| } |
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,48 @@ | ||
| // Copyright (c) Roman Atachiants and contributors. All rights reserved. | ||
| // Licensed under the MIT license. See LICENSE file in the project root for details. | ||
|
|
||
| package block | ||
|
|
||
| import ( | ||
| "io/ioutil" | ||
| "testing" | ||
|
|
||
| "github.com/kelindar/talaria/internal/encoding/typeof" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func setup() (Block, typeof.Schema) { | ||
|
|
||
| const testFile = "../../../test/test4.csv" | ||
| o, _ := ioutil.ReadFile(testFile) | ||
| schema := &typeof.Schema{ | ||
| "raisedCurrency": typeof.String, | ||
| "raisedAmt": typeof.Float64, | ||
| } | ||
| apply := Transform(schema) | ||
| b, _ := FromCSVBy(o, "raisedCurrency", &typeof.Schema{ | ||
| "raisedCurrency": typeof.String, | ||
| "raisedAmt": typeof.Float64, | ||
| }, apply) | ||
| if len(b) > 0 { | ||
| return b[0], *schema | ||
| } | ||
| return Block{}, *schema | ||
| } | ||
|
|
||
| func TestFromBlock(t *testing.T) { | ||
| blk, schema := setup() | ||
| rows, err := FromBlockBy(blk, schema) | ||
| assert.NoError(t, err) | ||
| cols, err := blk.Select(schema) | ||
| assert.NoError(t, err) | ||
| rowCount := cols.Any().Count() | ||
| // verify row count. | ||
| assert.Equal(t, rowCount, len(rows)) | ||
| for _, row := range rows { | ||
| // verify values | ||
| assert.Contains(t, []string{"EUR", "CAD", "USD"}, row.Values["raisedCurrency"]) | ||
| // verify type | ||
| assert.Equal(t, typeof.String, row.Schema["raisedCurrency"]) | ||
| } | ||
| } |
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
Oops, something went wrong.