[
https://issues.apache.org/jira/browse/ARROW-7109?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16973604#comment-16973604
]
Leo Meyerovich commented on ARROW-7109:
---------------------------------------
Awesome!
How to stream record batches in/out seems like the most meaningful. At least
here, a lot of our use ends up being about this. So maybe help to share some of
the arrow APIs used and maybe how/why.
# Maybe a start is revisiting the early microbatch tutorial I had done before
the API stabilized:
[https://observablehq.com/d/6e565d7662d984ea
|https://observablehq.com/d/6e565d7662d984ea]
^^^ Aimed to show microbatch API and prove some numbers
2. As a followup, demoing common fast IO needs: fast browser<>node, node
process <> node process, and node process <> python process. A lot of our code
ends up looking like:
```
export function asTable(source: Table | AsyncIterable<RecordBatch>, fields:
string[] = []) {
return AsyncIterableX.defer(async () => {
if (source) {
const batches = source instanceof Table ? source.chunks : await
toArray(source);
if (batches.length > 0) {
if (!fields || !fields.length) {
return AsyncIterableX.of(new Table(batches));
}
const table = new Table(batches).select(...fields);
if (table.schema.fields.length === fields.length) {
return AsyncIterableX.of(table);
}
}
}
return AsyncIterableX.empty<Table>();
});
}
export function asBatches(fn: () => DeferFnReturn) {
return asReaders(fn).concatAll() as BatchesReturn;
}
export function asReaders(fn: () => DeferFnReturn) {
return AsyncIterableX.defer(async () => {
const x = await fn();
return RecordBatchStreamReader.readAll(x);
}) as ReadersReturn;
}
```
and
```
asReaders(() => {
const nodeEncodingsStream = asTable(Array.isArray(nodeEncodings) ?
this.node.table.encodings(nodeEncodings) : nodeEncodings);
const edgeEncodingsStream = asTable(Array.isArray(edgeEncodings) ?
this.edge.table.encodings(edgeEncodings) : edgeEncodings);
const recordBatchesStream = AsyncIterable.concat(
// `preshaped` is an AsyncIterable<AsyncIterable<RecordBatch>>,
so flatten it here
AsyncIterable.as(preshaped).flatMap((x) => asTable(x)),
// Calling memoize() on the node/edge encodings streams ensures
they start downloading
// immediately, instead of serially after the incoming
RecordBatch streams have completed
nodeEncodingsStream.memoize().flatMap((\{ chunks }) =>
AsyncIterable.as(chunks)),
edgeEncodingsStream.memoize().flatMap((\{ chunks }) =>
AsyncIterable.as(chunks)),
);
return recordBatchesStream
.pipe(RecordBatchStreamWriter.throughNode(\{ autoDestroy: false
}))
.pipe(this.got.stream.post('/preshaped/shaped', \{ headers:
octetstream }));
});
```
> [JS] Create table from Arrays
> -----------------------------
>
> Key: ARROW-7109
> URL: https://issues.apache.org/jira/browse/ARROW-7109
> Project: Apache Arrow
> Issue Type: Wish
> Reporter: Sascha Hofmann
> Assignee: Brian Hulette
> Priority: Minor
> Attachments: image-2019-11-12-09-11-39-751.png
>
>
> I am trying to generate an arrow table from JS arrays and followed the
> example from [here |
> [https://observablehq.com/@lmeyerov/manipulating-flat-arrays-arrow-style]]
> but I am struggling to generate different schemas, most importantly how to
> provide a type different then 'floatingpoint'. Right now, I have:
> {code:java}
> const data = Table.from({
> schema: {
> fields: [{name: 'a', nullable: false, children: Array(0),
> type: {name: 'floatingpoint', precision: 'SINGLE'}}]
> },
> batches: [{
> count: 10,
> columns: [{ name: 'a', count: 10, VALIDITY: [],
> DATA: Array.from({ length: 10 }, () => 'a') }]}]
> })
> {code}
> Which, of course is non-sense but I couldn't figure out how to provide the
> type (I tried type: Utf8 among others). In general wouldn't it be a nice to
> have create Table from object function?
> On another note, are there any plans to make the docs a little bit more
> descriptive? Happy to contribute!
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
