infzo opened a new issue, #35703:
URL: https://github.com/apache/arrow/issues/35703
### Describe the usage question you have. Please include as many useful
details as possible.
**When I use do put, the server writes data, but the client can only read
none. Why?**
Logs:
```
# server
2023-05-22 12:15:57.491 | INFO | __main__:__init__:13 - server init:
grpc+tcp://localhost:5555
2023-05-22 12:16:18.571 | ERROR | __main__:do_put:25 - do put: cmd
b'xxx', data size: 10 4, 0 MB, column names: ['col_0', 'col_1', 'col_2',
'col_3']
2023-05-22 12:16:18.571 | ERROR | __main__:do_put:29 - write success
# client
create_table cost 0.0004131793975830078 s
result = None
result = None
result = None
result = None
result = None
result = None
result = None
```
Server Code:
```python
import struct
import pyarrow
from pyarrow import flight
from pyarrow.flight import FlightServerBase
from loguru import logger
class FlightRpcListener(FlightServerBase):
def __init__(self, host, port):
super().__init__(self._make_location(host, port))
logger.info(f'server init: {self._make_location(host, port)}')
@classmethod
def _make_location(cls, host: str, port: int) -> str:
return 'grpc+tcp://' + host + ':' + str(port)
def do_put(self, context, descriptor, reader, writer):
if descriptor.descriptor_type != flight.DescriptorType.CMD:
raise pyarrow.ArrowInvalid("Must provide a command descriptor")
cmd: str = descriptor.command
table: pyarrow.Table = reader.read_all()
logger.error(f'do put: cmd {cmd}, data size: {table.num_rows}
{table.num_columns}, '
f'{table.nbytes // (1024 ** 2)} MB, column names:
{table.column_names}')
writer.write(struct.pack('<i', 1000))
logger.error('write success')
server = FlightRpcListener(host='localhost', port=5555)
server.serve()
```
Client Code:
```
# -*- coding: UTF-8 -*-
import time
import numpy as np
import pyarrow
from pyarrow import flight
def create_table(n_rows, n_cols):
start = time.time()
table = pyarrow.Table.from_pydict({f'col_{c}': np.random.randint(0, 100,
size=[n_rows]) for c in range(n_cols)})
print(f'create_table cost {time.time() - start} s')
return table
def create_client(port):
return pyarrow.flight.FlightClient(f'grpc+tcp://localhost:{port}')
def run(cli, data):
writer, reader = cli.do_put(
flight.FlightDescriptor.for_command('xxx'),
data.schema)
writer.write_table(data)
writer.close()
while True:
print(f'result = {reader.read()}')
time.sleep(1)
def start_write_stream(cli):
data = create_table(10, 4)
run(cli, data)
if __name__ == '__main__':
client = create_client(5555)
start_write_stream(client)
```
### Component(s)
FlightRPC, Python
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
