TensorFlow Federated - Loading and preprocessing data on a remote client

Question

Part of the simulation program that I am working on allows clients to load local data from their device without the server being able to access that data.

Following the idea from this post, I have the following code configured to assign the client a path to load the data from. Although the data is in svmlight format, loading it line-by-line can still allow it to be preprocessed afterwards.

client_paths = {
    'client_0': '<path_here>',
    'client_1': '<path_here>',
}

def create_tf_dataset_for_client_fn(id):
    path = client_paths.get(id)
    data = tf.data.TextLineDataset(path) 

path_source = tff.simulation.datasets.ClientData.from_clients_and_fn(client_paths.keys(), create_tf_dataset_for_client_fn)

The code above allows a path to be loaded during runtime from the remote client's-side by the following line of code.

data = path_source.create_tf_dataset_for_client('client_0')

Here, the data variable can be iterated through and can be used to display the contents on the client on the remote device when calling tf.print(). But, I need to preprocess this data into an appropriate format before continuing. I am presently attempting to convert this from a string Tensor in svmlight format into a SparseTensor of the appropriate format.

The issue is that, although the defined preprocessing method works in a standalone scenario (i.e. when defined as a function and tested on a manually defined Tensor of the same format), it fails when the code is executed during the client update @tf.function in the tff algorithm. Below is the specified error when executing the notebook cell which contains a @tff.tf_computation function which calls an @tf.function which does the preprocessing and retrieves the data.

ValueError: Shape must be rank 1 but is rank 0 for '{{node Reshape_2}} = Reshape[T=DT_INT64, Tshape=DT_INT32](StringToNumber_1, Reshape_2/shape)' with input shapes: [?,?], [].

Since the issue occurs when executing the client's @tff.tf_computation update function which calls the @tf.function with the preprocessing code, I am wondering how I can allow the function to perform the preprocessing on the data without errors. I assume that if I can just get the functions to properly be run when defined that when called remotely it will work.

Any ideas on how to address this issue? Thank you for your help!

For reference, the preprocessing function uses tf computations to manipulate the data. Although not optimal yet, below is the code presently being used. This is inspired from this link on string_split examples. I have extracted the code to put directly into the client's @tf.function after loading the TextLineDataset as well, but this also fails.

def decode_libsvm(line):
        # Split the line into columns, delimiting by a blank space
        cols = tf.strings.split([line], ' ')
        # Retrieve the labels from the first column as an integer
        labels = tf.strings.to_number(cols.values[0], out_type=tf.int32)
        # Split all column pairs
        splits = tf.strings.split(cols.values[1:], ':')
        # Convert splits into a sparse matrix to retrieve all needed properties
        splits = splits.to_sparse()
        # Reshape the tensor for further processing
        id_vals = tf.reshape(splits.values, splits.dense_shape)
        # Retrieve the indices and values within two separate tensors
        feat_ids, feat_vals = tf.split(id_vals, num_or_size_splits=2, axis=1)
        # Convert the indices into int64 numbers
        feat_ids = tf.strings.to_number(feat_ids, out_type=tf.int64)
        # To reload within a SparseTensor, add a dimension to feat_ids with a default value of 0
        feat_ids = tf.reshape(feat_ids, -1)
        feat_ids = tf.expand_dims(feat_ids, 1)
        feat_ids = tf.pad(feat_ids, [[0,0], [0,1]], constant_values=0)
        # Extract and flatten the values
        feat_vals = tf.strings.to_number(feat_vals, out_type=tf.float32)
        feat_vals = tf.reshape(feat_vals, -1)
        # Configure a SparseTensor to contain the indices and values
        sparse_output = tf.SparseTensor(indices=feat_ids, values=feat_vals, dense_shape=[1, <shape>])
        return {"x": sparse_output, "y": labels}

Update (Fix)

Following the advice from Jakub's comment, the issue was fixed by enclosing the reshape and expand_dim calls in [], when needed. Now there is no issue running the code within tff.

def decode_libsvm(line):
        # Split the line into columns, delimiting by a blank space
        cols = tf.strings.split([line], ' ')
        # Retrieve the labels from the first column as an integer
        labels = tf.strings.to_number(cols.values[0], out_type=tf.int32)
        # Split all column pairs
        splits = tf.strings.split(cols.values[1:], ':')
        # Convert splits into a sparse matrix to retrieve all needed properties
        splits = splits.to_sparse()
        # Reshape the tensor for further processing
        id_vals = tf.reshape(splits.values, splits.dense_shape)
        # Retrieve the indices and values within two separate tensors
        feat_ids, feat_vals = tf.split(id_vals, num_or_size_splits=2, axis=1)
        # Convert the indices into int64 numbers
        feat_ids = tf.strings.to_number(feat_ids, out_type=tf.int64)
        # To reload within a SparseTensor, add a dimension to feat_ids with a default value of 0
        feat_ids = tf.reshape(feat_ids, [-1])
        feat_ids = tf.expand_dims(feat_ids, [1])
        feat_ids = tf.pad(feat_ids, [[0,0], [0,1]], constant_values=0)
        # Extract and flatten the values
        feat_vals = tf.strings.to_number(feat_vals, out_type=tf.float32)
        feat_vals = tf.reshape(feat_vals, [-1])
        # Configure a SparseTensor to contain the indices and values
        sparse_output = tf.SparseTensor(indices=feat_ids, values=feat_vals, dense_shape=[1, <shape>])
        return {"x": sparse_output, "y": labels}