The Streams Python development guide mentions how you can quickly read local files from a Python application.  However, what if the file is in a project in Cloud Pak for Data? How can you access files in a project from a Streams notebook?
In this post I’ll cover how you can use data from a file in a Streams Python notebook running in IBM Cloud Pak for Data.


Note for other languages: If you are developing with SPL, working with local files and directories is easy to do with the FileSource and DirectoryScan operators. Applications running in the IBM Cloud can apply the cloud-readiness techniques discussed on this page. The Streams Java API technique is discussed here.

How to do it

Step 1: Add the file to the project

Optional: Find the path to the file relative to the notebook

Step 2: Include the file in the Streams application

• Include the file in the code bundle that is sent to the Streams instance using Topology.add_file_dependency().
• Use streamsx.ec.get_application_directory() to get the path to the file when you are ready to read it.

Putting it together

import streamsx.ec
class CSVFileReader:
    def __init__(self, file): # this is called in the notebook
        self.file_name = file
    def __call__(self): # called at runtime on the Streams instance
       # open the file
        with open(streamsx.ec.get_application_directory()
               + "/etc/" + self.file_name) as handle:

Complete example

#ticker, date, time, askprice
"AEC","27-MAR-2020","06:16:53.068",0
"AGU","27-MAR-2020","06:16:53.068",0
"AOF","27-MAR-2020","06:16:53.068",0
"AWF","27-MAR-2020","06:16:54.074",0

from streamsx.topology.topology import Topology
from icpd_core import icpd_util
import streamsx.topology.context
import os
import csv
import streamsx.ec

# The CSVFileReader class opens the file at runtime and generates tuples:

class CSVFileReader:
    def __init__(self, file):
        self.file_name = file
    def __call__(self):
        # Convert each row in the file to a dict
        header = ["ticker","date", "time", "askprice"]
        with open(streamsx.ec.get_application_directory()
               + "/etc/" + self.file_name) as handle:
            reader = csv.DictReader(handle, delimiter=',',
                                    fieldnames=header)
            #Use this to skip the header line if your file has one
            next(reader)
            for row in reader:
                yield row  # submit each row  


file = "trades.csv"
topo = Topology("AnalyzeData") # add the file as a dependency 
topo.add_file_dependency('/project_data/data_asset/' + file, "etc")

#use the CSVFileReader class to open the file on the Streams runtime host 
lines_from_file = topo.source(CSVFileReader(file))
lines_from_file.filter(lambda tpl: float(tpl["askprice"]) > 0).print() 

streams_instance_name = "sample-streams" ## Change this to Streams instance
try:
    cfg=icpd_util.get_service_instance_details(name=streams_instance_name, instance_type="streams")
except TypeError:
    cfg=icpd_util.get_service_instance_details(name=streams_instance_name)

# Disable SSL certificate verification if necessary
cfg[context.ConfigParams.SSL_VERIFY] = False
contextType = context.ContextTypes.DISTRIBUTED
context.submit (contextType, topo, config = cfg)