-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #29 from bytewax/metrics
Adding kube-prometheus-stack and some fixes
- Loading branch information
Showing
14 changed files
with
284 additions
and
107 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,12 @@ | ||
| dependencies: | ||
| - name: opentelemetry-collector | ||
| repository: https://open-telemetry.github.io/opentelemetry-helm-charts | ||
| version: 0.36.2 | ||
| version: 0.36.3 | ||
| - name: jaeger | ||
| repository: https://jaegertracing.github.io/helm-charts | ||
| version: 0.62.1 | ||
| digest: sha256:e802d990a09bd88e4ea4f30126c25439ffea95ec5c70af4c4eaf76f4eeece3b5 | ||
| generated: "2022-10-17T15:11:11.549316385-03:00" | ||
| - name: kube-prometheus-stack | ||
| repository: https://prometheus-community.github.io/helm-charts | ||
| version: 56.2.1 | ||
| digest: sha256:c930a7e7afc57cb42aa0d2cc589d1523b910e7e83cc6b0e445130d5689bac272 | ||
| generated: "2024-01-30T08:27:06.827785793-03:00" |
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
Binary file not shown.
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,29 +1,32 @@ | ||
| import bytewax.operators as op | ||
| from bytewax.testing import TestingSource | ||
| from bytewax.dataflow import Dataflow | ||
| from bytewax.connectors.stdio import StdOutput | ||
| from bytewax.inputs import StatelessSource, DynamicInput | ||
| from bytewax.connectors.stdio import StdOutSink | ||
| from bytewax.inputs import ( | ||
| DynamicSource, | ||
| StatelessSourcePartition, | ||
| ) | ||
| import time | ||
|
|
||
| class NumberSource(StatelessSource): | ||
| class NumberSource(StatelessSourcePartition): | ||
| def __init__(self, max, worker_index): | ||
| self.worker_index = worker_index | ||
| self.iterator = iter(range(max)) | ||
|
|
||
| def next(self): | ||
| def next_batch(self, worker_index): | ||
| time.sleep(1) | ||
| return f"Worker: {self.worker_index} - {next(self.iterator)}" | ||
| return [f"Worker: {self.worker_index} - {next(self.iterator)}"] | ||
|
|
||
| def close(self): | ||
| pass | ||
|
|
||
|
|
||
| class NumberInput(DynamicInput): | ||
| class NumberInput(DynamicSource): | ||
| def __init__(self, max): | ||
| self.max = max | ||
|
|
||
| def build(self, worker_index, worker_count): | ||
| return NumberSource(self.max, worker_index) | ||
|
|
||
| def build(self, _now, worker_index, worker_count): | ||
| return NumberSource(max=self.max, worker_index=worker_index) | ||
|
|
||
| flow = Dataflow() | ||
| flow.input("inp", NumberInput(100)) | ||
| flow.output("out", StdOutput()) | ||
| flow = Dataflow("k8s_basic") | ||
| out = op.input("inp1", flow, NumberInput(100)) | ||
| op.output("out", out, StdOutSink()) |
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,47 +1,66 @@ | ||
| import os | ||
| from pathlib import Path | ||
|
|
||
| from bytewax.dataflow import Dataflow | ||
| from bytewax.connectors.stdio import StdOutput | ||
| from bytewax.connectors.files import DirInput, DirOutput, FileInput, FileOutput | ||
|
|
||
| input_dir = Path("./sample_data/cluster/") | ||
| output_dir = Path("./cluster_out/") | ||
| import bytewax.operators as op | ||
| from bytewax.connectors.stdio import StdOutSink | ||
| from bytewax.connectors.files import ( | ||
| DirSink, | ||
| DirSource, | ||
| ) | ||
|
|
||
| def to_tuple(x): | ||
| return tuple(map(str, x.split(','))) | ||
|
|
||
| flow = Dataflow() | ||
| flow.input("inp", DirInput(input_dir)) | ||
| flow.map(str.upper) | ||
| flow.map(to_tuple) | ||
| flow.output("out", DirOutput(output_dir, 5, assign_file=int)) | ||
| flow = Dataflow("k8s_cluster") | ||
| inp1 = op.input("inp", flow, DirSource(Path("./sample_data/cluster"))) | ||
| inp2 = op.map("upper", inp1, str.upper) | ||
| out = op.map("tuple", inp2, to_tuple) | ||
| op.output("out1", out, DirSink(Path("./cluster_out/"), 2, assign_file=int)) | ||
| op.output("out2", out, StdOutSink()) | ||
|
|
||
|
|
||
| # import os | ||
| # from pathlib import Path | ||
|
|
||
| # from bytewax.dataflow import Dataflow | ||
| # from bytewax.connectors.stdio import StdOutput | ||
| # from bytewax.connectors.files import DirInput, DirOutput, FileInput, FileOutput | ||
|
|
||
| # input_dir = Path("./sample_data/cluster/") | ||
| # output_dir = Path("./cluster_out/") | ||
|
|
||
| # def to_tuple(x): | ||
| # return tuple(map(str, x.split(','))) | ||
|
|
||
| # flow = Dataflow() | ||
| # flow.input("inp", DirInput(input_dir)) | ||
| # flow.map(str.upper) | ||
| # flow.map(to_tuple) | ||
| # flow.output("out", DirOutput(output_dir, 5, assign_file=int)) | ||
|
|
||
|
|
||
| # We are going to use Waxctl, you can download it from https://bytewax.io/downloads | ||
| # Run these commands in your terminal to run a cluster of two containers: | ||
|
|
||
| # $ tar -C ./ -cvf cluster.tar examples | ||
| # $ waxctl dataflow deploy ./cluster.tar --name k8s-cluster --python-file-name examples/k8s_cluster.py -p2 | ||
| # $ waxctl dataflow deploy ./cluster.tar \ | ||
| # --name k8s-cluster \ | ||
| # --python-file-name examples/k8s_cluster.py \ | ||
| # -p2 --debug --keep-alive=true --yes | ||
|
|
||
| # Each worker will read the files in | ||
| # ./examples/sample_data/cluster/*.txt which have lines like | ||
| # `one1`. | ||
| # `ONE1`. | ||
|
|
||
| # They will then both finish and you'll see ./cluster_out/part_0 | ||
| # and ./cluster_out/part_1 with the data that each process in the | ||
| # cluster wrote with the lines uppercased. | ||
|
|
||
| # To see that files in each container you can run these commands: | ||
|
|
||
| # kubectl exec -it k8s-cluster-0 -cprocess -- cat /var/bytewax/cluster_out/part_0.out | ||
| # kubectl exec -it k8s-cluster-1 -cprocess -- cat /var/bytewax/cluster_out/part_1.out | ||
| # kubectl exec -it k8s-cluster-0 -cprocess -- cat /var/bytewax/cluster_out/part_0 | ||
| # kubectl exec -it k8s-cluster-1 -cprocess -- cat /var/bytewax/cluster_out/part_1 | ||
|
|
||
| # You could imagine reading from / writing to separate Kafka | ||
| # partitions, S3 blobs, etc. | ||
|
|
||
| # When using `cluster_main()` you have to coordinate ensuring each | ||
| # process knows the address of all other processes in the cluster | ||
| # and their unique process ID. You can address that easily by deploying your | ||
| # dataflow program using Waxctl or installing the Bytewax Helm Chart | ||
| # cluster_main(flow, recovery_config=recovery_config, **parse.proc_env()) |
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,9 +1,10 @@ | ||
| # ./simple.py | ||
| import bytewax.operators as op | ||
| from bytewax.testing import TestingSource | ||
| from bytewax.dataflow import Dataflow | ||
| from bytewax.testing import TestingInput | ||
| from bytewax.connectors.stdio import StdOutput | ||
| from bytewax.connectors.stdio import StdOutSink | ||
|
|
||
| flow = Dataflow() | ||
| flow.input("inp", TestingInput(range(99999999))) | ||
| flow.map(lambda item: item + 1) | ||
| flow.output("out", StdOutput()) | ||
| flow = Dataflow("simple") | ||
|
|
||
| out = op.input("inp1", flow, TestingSource(range(99999999))) | ||
| op.output("out", out, StdOutSink()) |
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,14 +1,20 @@ | ||
| # ./simple.py | ||
| import bytewax.operators as op | ||
| from bytewax.testing import TestingSource | ||
| from bytewax.dataflow import Dataflow | ||
| from bytewax.testing import TestingInput | ||
| from bytewax.connectors.stdio import StdOutput | ||
| from bytewax.connectors.stdio import StdOutSink | ||
| import time | ||
|
|
||
| def slow_inc(x): | ||
| time.sleep(5) | ||
| return x + 1 | ||
|
|
||
| flow = Dataflow() | ||
| flow.input("inp", TestingInput(range(99999999))) | ||
| flow.map(slow_inc) | ||
| flow.output("out", StdOutput()) | ||
| flow = Dataflow("simple") | ||
|
|
||
| inp = op.input("inp1", flow, TestingSource(range(99999999))) | ||
| out = op.map("slow", inp, slow_inc) | ||
| op.output("out", out, StdOutSink()) | ||
|
|
||
|
|
||
|
|
||
|
|
Oops, something went wrong.