changes to oci streaming

data-platform/open-source-data-platforms/oci-streaming/README.md

@@ -11,7 +11,6 @@ Reviewed: 04.06.2024
 
 # Team Publications
 
-- [Revival of the Magical Suitcase](https://medium.com/@devpiotrekk/revival-of-the-magical-suitcase-73093af23f29)
 - [LiveLabs: Deploying OCI Streaming Service](https://apexapps.oracle.com/pls/apex/f?p=133:180:107188281482541::::wid:664)
 - [LiveLabs: How do I ingest messages into OCI streaming in real-time with Oracle GoldenGate for Big Data?](https://apexapps.oracle.com/pls/apex/r/dbpm/livelabs/run-workshop?p210_wid=3572&session=107188281482541)
 
@@ -20,6 +19,13 @@ Reviewed: 04.06.2024
 - [OCI Streaming Documentation](https://docs.oracle.com/en-us/iaas/Content/Streaming/Concepts/streamingoverview.htm)
 - [Overview of OCI Streaming - YouTube](https://www.youtube.com/watch?v=G8-E_j-uVak)
 
+# Reusable Assets
+
+- [Revival of the Magical Suitcase](https://medium.com/@devpiotrekk/revival-of-the-magical-suitcase-73093af23f29)
+
+- [Fake Producer and Consumer for OCI Streaming](https://github.com/oracle-devrel/technology-engineering/tree/main/data-platform/open-source-data-platforms/oci-streaming/code-examples/fake-producer-consumer)
+
+- [Create and run Mosquitto & Node-RED, connecting to OCI Streaming](https://github.com/oracle-devrel/technology-engineering/tree/main/data-platform/open-source-data-platforms/oci-streaming/code-examples/mosquitto_node-red)
 
 # License
 

data-platform/open-source-data-platforms/oci-streaming/code-examples/fake-producer-consumer/LICENSE

@@ -0,0 +1,35 @@
+Copyright (c) 2024 Oracle and/or its affiliates.
+
+The Universal Permissive License (UPL), Version 1.0
+
+Subject to the condition set forth below, permission is hereby granted to any
+person obtaining a copy of this software, associated documentation and/or data
+(collectively the "Software"), free of charge and under any and all copyright
+rights in the Software, and any and all patent rights owned or freely
+licensable by each licensor hereunder covering either (i) the unmodified
+Software as contributed to or provided by such licensor, or (ii) the Larger
+Works (as defined below), to deal in both
+
+(a) the Software, and
+(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if
+one is included with the Software (each a "Larger Work" to which the Software
+is contributed by such licensors),
+
+without restriction, including without limitation the rights to copy, create
+derivative works of, display, perform, and distribute the Software and make,
+use, sell, offer for sale, import, export, have made, and have sold the
+Software and the Larger Work(s), and to sublicense the foregoing rights on
+either these or other terms.
+
+This license is subject to the following condition:
+The above copyright notice and either this complete permission notice or at
+a minimum a reference to the UPL must be included in all copies or
+substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

data-platform/open-source-data-platforms/oci-streaming/code-examples/fake-producer-consumer/README.md

@@ -0,0 +1,42 @@
+# Example of Producing and Consuming for OCI Streaming
+
+Reviewed: 22.10.2024
+
+1. Create compute instance. Oracle Linux 7.
+2. Run the below to install Git, clone the repo, and install several packages
+  ```
+  sudo dnf install git-all -y
+  git clone https://github.com/bobpeulen/apache_kafka.git
+  sudo pip3 install kafka-python3 pandas numpy datetime
+  ```
+
+3. Run the producer. 
+
+```
+python apache_kafka/oci_streaming/producer.py  \
+-tenancy_name 'oraemeadatamgmt' \
+-region 'eu-frankfurt-1' \
+-user_name 'OracleIdentityCloudService/[email protected]' \
+-stream_name 'OpenSourceData_stream_1' \
+-stream_pool_ocid 'ocid1.streampool.oc1.eu-frankfurt-1.amaaaaaaeicj2tiacazj6xzvn7rkfdyci6w2io634erapt7ctpxtqxauvocmea' \
+-auth_token 'ADD YOUR TOKEN HERE'
+```
+
+4. Run the consumer.
+ ```
+  python apache_kafka/oci_streaming/consumer.py  \
+  -tenancy_name 'oraemeadatamgmt' \
+  -region 'eu-frankfurt-1' \
+  -user_name 'OracleIdentityCloudService/[email protected]' \
+  -stream_name 'OpenSourceData_stream_1' \
+  -stream_pool_ocid 'ocid1.streampool.oc1.eu-frankfurt-1.amaaaaaaeicj2tiacazj6xzvn7rkfdyci6w2io634erapt7ctpxtqxauvocmea' \
+  -auth_token 'ADD YOUR TOKEN HERE'
+ ```
+
+# License
+
+Copyright (c) 2024 Oracle and/or its affiliates.
+
+Licensed under the Universal Permissive License (UPL), Version 1.0.
+
+See [LICENSE](https://github.com/oracle-devrel/technology-engineering/blob/main/LICENSE) for more details.

data-platform/open-source-data-platforms/oci-streaming/code-examples/fake-producer-consumer/files/consumer.py

@@ -0,0 +1,42 @@
+from kafka3 import KafkaConsumer, KafkaProducer
+import pandas as pd
+from time import sleep
+import datetime
+import numpy as np
+from datetime import datetime, timedelta
+import random
+import argparse
+import uuid
+
+#get arguments
+parser = argparse.ArgumentParser()
+
+parser.add_argument("-tenancy_name", "--tenancy_name", help="tenancy_name", required=True)
+parser.add_argument("-region", "--region", help="region",required=True)
+parser.add_argument("-user_name", "--user_name", help="user_name including OracleIdentityCloudService", required=True)
+parser.add_argument("-stream_name", "--stream_name", help="stream_name / topic", required=True)
+parser.add_argument("-stream_pool_ocid", "--stream_pool_ocid", help="stream_pool_ocid", required=True)
+parser.add_argument("-auth_token", "--auth_token", help="auth_token", required=True)
+
+
+args = parser.parse_args()
+tenancy_name = args.tenancy_name
+region = args.region
+user_name = args.user_name
+stream_name = args.stream_name
+stream_pool_ocid = args.stream_pool_ocid
+auth_token = args.auth_token
+
+
+def main():
+  consumer = KafkaConsumer(stream_name, bootstrap_servers = f'cell-1.streaming.{region}.oci.oraclecloud.com:9092', 
+                         security_protocol = 'SASL_SSL', sasl_mechanism = 'PLAIN',
+                         sasl_plain_username = f'{tenancy_name}/{user_name}/{stream_pool_ocid}',
+                         sasl_plain_password = auth_token)
+
+  for message in consumer:
+    print(f"{message.key}: {message.value}")
+
+
+if __name__ == '__main__':
+    main()

data-platform/open-source-data-platforms/oci-streaming/code-examples/fake-producer-consumer/files/producer.py

@@ -0,0 +1,70 @@
+from kafka3 import KafkaConsumer, KafkaProducer
+import pandas as pd
+from time import sleep
+import datetime
+import numpy as np
+from datetime import datetime, timedelta
+import random
+import argparse
+import uuid
+
+#get arguments
+parser = argparse.ArgumentParser()
+
+parser.add_argument("-tenancy_name", "--tenancy_name", help="tenancy_name", required=True)
+parser.add_argument("-region", "--region", help="region",required=True)
+parser.add_argument("-user_name", "--user_name", help="user_name including OracleIdentityCloudService", required=True)
+parser.add_argument("-stream_name", "--stream_name", help="stream_name / topic", required=True)
+parser.add_argument("-stream_pool_ocid", "--stream_pool_ocid", help="stream_pool_ocid", required=True)
+parser.add_argument("-auth_token", "--auth_token", help="auth_token", required=True)
+
+
+args = parser.parse_args()
+tenancy_name = args.tenancy_name
+region = args.region
+user_name = args.user_name
+stream_name = args.stream_name
+stream_pool_ocid = args.stream_pool_ocid
+auth_token = args.auth_token
+
+
+def main():
+  ## create connection
+    producer = KafkaProducer(bootstrap_servers = f'cell-1.streaming.{region}.oci.oraclecloud.com:9092', linger_ms = 50, batch_size  = 2,
+                         security_protocol = 'SASL_SSL', sasl_mechanism = 'PLAIN',
+                         value_serializer = lambda v: v.encode('utf-8'),
+                         sasl_plain_username = f'{tenancy_name}/{user_name}/{stream_pool_ocid}',
+                         sasl_plain_password = auth_token)
+  #create random data
+    pd_list = []
+
+    for i in range(500):
+
+        #set randomness, so each goes up and down together
+        randomness_sun_price = random.uniform(0.95, 1.015)
+        randomness_wspd = random.uniform(0.90, 0.95)
+        cust_id = str(uuid.uuid4())
+        kwh_producing = round(random.uniform(0.11, 5.55), 2)
+        kwh_consuming = round(random.uniform(3.55, 19.55), 2)
+
+        ##weather info 
+        temp = randomness_sun_price
+        wspd =  randomness_wspd
+        tsun = randomness_sun_price *1.22 
+        energy_price = randomness_sun_price *0.89
+
+        pd_list.append([cust_id, kwh_producing, kwh_consuming, temp, wspd, tsun, energy_price])
+
+    df_out = pd.DataFrame(pd_list, columns = ["cust_id", "kwh_producing", "kwh_consuming", "temp", "windspeed","sun_light", "energy_price"])
+
+  #push random data to stream
+    for index, row in df_out.iterrows():
+        row_json = row.to_json(orient='records', lines=True)
+        producer.send(stream_name, value=row_json)
+        producer.flush()
+        print(f"Producing to stream: {row_json}")
+        sleep(5)
+
+
+if __name__ == '__main__':
+    main()

data-platform/open-source-data-platforms/oci-streaming/code-examples/fake-producer-consumer/files/producer_csv.py

@@ -0,0 +1,52 @@
+from kafka3 import KafkaConsumer, KafkaProducer
+import pandas as pd
+from time import sleep
+import datetime
+import numpy as np
+from datetime import datetime, timedelta
+import random
+import argparse
+import uuid
+
+#get arguments
+parser = argparse.ArgumentParser()
+
+parser.add_argument("-tenancy_name", "--tenancy_name", help="tenancy_name", required=True)
+parser.add_argument("-region", "--region", help="region",required=True)
+parser.add_argument("-user_name", "--user_name", help="user_name including OracleIdentityCloudService", required=True)
+parser.add_argument("-stream_name", "--stream_name", help="stream_name / topic", required=True)
+parser.add_argument("-stream_pool_ocid", "--stream_pool_ocid", help="stream_pool_ocid", required=True)
+parser.add_argument("-auth_token", "--auth_token", help="auth_token", required=True)
+
+
+args = parser.parse_args()
+tenancy_name = args.tenancy_name
+region = args.region
+user_name = args.user_name
+stream_name = args.stream_name
+stream_pool_ocid = args.stream_pool_ocid
+auth_token = args.auth_token
+
+
+def main():
+  ## create connection
+    producer = KafkaProducer(bootstrap_servers = f'cell-1.streaming.{region}.oci.oraclecloud.com:9092', linger_ms = 50, batch_size  = 2,
+                         security_protocol = 'SASL_SSL', sasl_mechanism = 'PLAIN',
+                         value_serializer = lambda v: v.encode('utf-8'),
+                         sasl_plain_username = f'{tenancy_name}/{user_name}/{stream_pool_ocid}',
+                         sasl_plain_password = auth_token)
+
+
+    df_out = pd.read_csv("./NAME_OF_CSV.csv")
+
+  #push random data to stream
+    for index, row in df_out.iterrows():
+        row_json = row.to_json(orient='records', lines=True)
+        producer.send(stream_name, value=row_json)
+        producer.flush()
+        print(f"Producing to stream: {row_json}")
+        sleep(5)
+
+
+if __name__ == '__main__':
+    main()

data-platform/open-source-data-platforms/oci-streaming/code-examples/mosquitto_node-red/LICENSE

@@ -0,0 +1,35 @@
+Copyright (c) 2024 Oracle and/or its affiliates.
+
+The Universal Permissive License (UPL), Version 1.0
+
+Subject to the condition set forth below, permission is hereby granted to any
+person obtaining a copy of this software, associated documentation and/or data
+(collectively the "Software"), free of charge and under any and all copyright
+rights in the Software, and any and all patent rights owned or freely
+licensable by each licensor hereunder covering either (i) the unmodified
+Software as contributed to or provided by such licensor, or (ii) the Larger
+Works (as defined below), to deal in both
+
+(a) the Software, and
+(b) any piece of software and/or hardware listed in the lrgrwrks.txt file if
+one is included with the Software (each a "Larger Work" to which the Software
+is contributed by such licensors),
+
+without restriction, including without limitation the rights to copy, create
+derivative works of, display, perform, and distribute the Software and make,
+use, sell, offer for sale, import, export, have made, and have sold the
+Software and the Larger Work(s), and to sublicense the foregoing rights on
+either these or other terms.
+
+This license is subject to the following condition:
+The above copyright notice and either this complete permission notice or at
+a minimum a reference to the UPL must be included in all copies or
+substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.