plasma on kubernetes

01_plasma.yaml

# Setup (on whatever worker the pods are running on):
#   mkdir /var/plasma-remi
#   mkdir /var/plasma-remi/socket
#   mkdir /var/plasma-remi/hugepages
#   mount -t hugetlbfs -o uid=0 -o gid=0 none /var/plasma-remi/hugepages

# Apply this config (kubectl apply -f plasma-remi.yaml)-test

# Execute plasma commands on client 1:
#   POD=$(kubectl get pod -l app=plasma-test-remi,thing=client1 -o='jsonpath={.items[*].metadata.name}')
#   kubectl exec -ti $POD python
#   > from pyarrow import plasma
#   > client = plasma.connect('/mnt/socket/plasma', '', 0)
#   > client.put("sample message here").binary

# Execute commands on client 2:
#   POD=$(kubectl get pod -l app=plasma-test-remi,thing=client2 -o='jsonpath={.items[*].metadata.name}')
#   > from pyarrow import plasma
#   > client = plasma.connect('/mnt/socket/plasma', '', 0)
#   > client.get(plasma.ObjectID(b'...'))


apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: plasma-test-server
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: plasma-test-remi
        thing: server
    spec:
      containers:
      - name: main
        image: remram/python3-pyarrow
#        command: ['/bin/sh', '-c', 'plasma_store -s /mnt/socket/plasma -m 10000000']
# or
        command: ['/bin/sh', '-c', 'plasma_store -s /mnt/socket/plasma -m 1000000000 -d /mnt/hugepages -h']
        volumeMounts:
        - mountPath: /mnt/socket
          name: socket
        - mountPath: /mnt/hugepages
          name: hugepages
      volumes:
      - name: socket
        hostPath:
          path: "/var/plasma-remi/socket"
      - name: hugepages
        hostPath:
          path: "/var/plasma-remi/hugepages"
---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: plasma-test-client1
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: plasma-test-remi
        thing: client1
    spec:
      containers:
      - name: main
        image: remram/python3-pyarrow
        command: ['/bin/sh', '-c', 'while true; do sleep 30; done']
        volumeMounts:
        - mountPath: /mnt/socket
          name: socket
      volumes:
      - name: socket
        hostPath:
          path: "/var/plasma-remi/socket"
---
apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  name: plasma-test-client2
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: plasma-test-remi
        thing: client2
    spec:
      containers:
      - name: main
        image: remram/python3-pyarrow
        command: ['/bin/sh', '-c', 'while true; do sleep 30; done']
        volumeMounts:
        - mountPath: /mnt/socket
          name: socket
      volumes:
      - name: socket
        hostPath:
          path: "/var/plasma-remi/socket"

02_writer.py

from pyarrow import plasma

client = plasma.connect('/mnt/socket/plasma', '', 0)

object_id = plasma.ObjectID(b'\x80\x04R\x8fR\xef\x1d\x01\xb3mC\x85\xe43\x1c16Q\xfe\xd5')

import numpy as np
import pyarrow as pa

# Create a pyarrow.Tensor object from a numpy random 2-dimensional array
data = np.random.randn(10, 4)
tensor = pa.Tensor.from_numpy(data)

# Create the object in Plasma
data_size = pa.get_tensor_size(tensor)
buf = client.create(object_id, data_size)

# Write the tensor into the Plasma-allocated buffer
stream = pa.FixedSizeBufferWriter(buf)
pa.write_tensor(tensor, stream)  # Writes tensor's 552 bytes to Plasma stream

# Seal the Plasma object
client.seal(object_id)

03_reader.py

from pyarrow import plasma

client = plasma.connect('/mnt/socket/plasma', '', 0)

object_id = plasma.ObjectID(b'\x80\x04R\x8fR\xef\x1d\x01\xb3mC\x85\xe43\x1c16Q\xfe\xd5')

import numpy as np
import pyarrow as pa

# Get the arrow object by ObjectID.
[buf2] = client.get_buffers([object_id])

# Reconstruct the Arrow tensor object.
reader = pa.BufferReader(buf2)
tensor2 = pa.read_tensor(reader)

# Convert back to numpy
array = tensor2.to_numpy()