Skip to content

Add Apache Arrow dataset support #36

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Jan 18, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,8 @@ sh_binary(
"LICENSE",
"MANIFEST.in",
"setup.py",
"tensorflow_io/__init__.py",
"tensorflow_io/__init__.py",
"//tensorflow_io/arrow:arrow_py",
"//tensorflow_io/hadoop:hadoop_py",
"//tensorflow_io/ignite:ignite_py",
"//tensorflow_io/kafka:kafka_py",
Expand Down
3 changes: 2 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,12 @@
TensorFlow I/O is a collection of file systems and file formats that are not
available in TensorFlow's built-in support.

At the moment TensorFlow I/O supports 4 data sources:
At the moment TensorFlow I/O supports 5 data sources:
- `tensorflow_io.ignite`: Data source for Apache Ignite and Ignite File System (IGFS).
- `tensorflow_io.kafka`: Apache Kafka stream-processing support.
- `tensorflow_io.kinesis`: Amazon Kinesis data streams support.
- `tensorflow_io.hadoop`: Hadoop SequenceFile format support.
- `tensorflow_io.arrow`: Apache Arrow data format support. Usage guide [here](tensorflow_io/arrow/README.md).

## Installation

Expand Down
56 changes: 56 additions & 0 deletions tensorflow_io/arrow/BUILD
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
licenses(["notice"]) # Apache 2.0

package(default_visibility = ["//visibility:public"])

cc_binary(
name = 'python/ops/_arrow_ops.so',
srcs = [
"kernels/arrow_dataset_ops.cc",
"kernels/arrow_stream_client.h",
"kernels/arrow_stream_client_unix.cc",
"ops/dataset_ops.cc",
],
linkshared = 1,
deps = [
"@local_config_tf//:libtensorflow_framework",
"@local_config_tf//:tf_header_lib",
"@arrow//:arrow",
],
copts = ["-pthread", "-std=c++11", "-D_GLIBCXX_USE_CXX11_ABI=0", "-DNDEBUG"]
)

py_library(
name = "arrow_ops_py",
srcs = [
"python/ops/arrow_dataset_ops.py",
],
data = [
":python/ops/_arrow_ops.so"
],
srcs_version = "PY2AND3",
)

py_test(
name = "arrow_py_test",
srcs = [
"python/kernel_tests/arrow_test.py"
],
main = "python/kernel_tests/arrow_test.py",
deps = [
":arrow_ops_py"
],
srcs_version = "PY2AND3",
)

py_library(
name = "arrow_py",
srcs = [
"__init__.py",
"python/__init__.py",
"python/ops/__init__.py",
],
deps = [
":arrow_ops_py",
],
srcs_version = "PY2AND3",
)
111 changes: 111 additions & 0 deletions tensorflow_io/arrow/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
# TensorFlow I/O Apache Arrow Datasets

Apache Arrow is a standard for in-memory columnar data, see [here](https://arrow.apache.org)
for more information on the project. An Arrow dataset makes it easy to bring in
column-oriented data from other systems to TensorFlow using the following
sources:

## From a Pandas DataFrame

An `ArrowDataset` can be made directly from an existing Pandas DataFrame, or
pyarrow record batches, in a Python process. Tensor types and shapes can be
inferred from the DataFrame, although currently only scalar and vector values
with primitive types are supported. PyArrow must be installed to use this
Dataset. Example usage:

```python
import tensorflow as tf
from tensorflow_io.arrow import ArrowDataset

# Assume `df` is an existing Pandas DataFrame
dataset = ArrowDataset.from_pandas(df)

iterator = dataset.make_one_shot_iterator()
next_element = iterator.get_next()

with tf.Session() as sess:
for i in range(len(df)):
print(sess.run(next_element))
```

NOTE: The entire DataFrame will be serialized to the Dataset and is not
recommended for use with large amounts of data

## From Arrow Feather Files

Feather is a light-weight file format that provides a simple and efficient way
to write Pandas DataFrames to disk, see [here](https://arrow.apache.org/docs/python/ipc.html#feather-format)
for more information and limitations of the format. An `ArrowFeatherDataset`
can be created to read one or more Feather files from the given pathnames. The
following example shows how to write a feather file from a Pandas DataFrame,
then read multiple files back as an `ArrowFeatherDataset`:

```python
from pyarrow.feather import write_feather

# Assume `df` is an existing Pandas DataFrame with dtypes=(int32, float32)
write_feather(df, '/path/to/a.feather')
```

```python
import tensorflow as tf
from tensorflow_io.arrow import ArrowFeatherDataset

# Each Feather file must have the same column types, here we use the above
# DataFrame which has 2 columns with dtypes=(int32, float32)
dataset = ArrowFeatherDataset(
['/path/to/a.feather', '/path/to/b.feather'],
columns=(0, 1),
output_types=(tf.int32, tf.float32),
output_shapes=([], []))

iterator = dataset.make_one_shot_iterator()
next_element = iterator.get_next()

# This will iterate over each row of each file provided
with tf.Session() as sess:
while True:
try:
print(sess.run(next_element))
except tf.errors.OutOfRangeError:
break
```

An alternate constructor can also be used to infer output types and shapes from
a given `pyarrow.Schema`, e.g. `dataset = ArrowFeatherDataset.from_schema(filenames, schema)`

## From a Stream of Arrow Record Batches

The `ArrowStreamDataset` provides a Dataset that will connect to a host over
a socket that is serving Arrow record batches in the Arrow stream format. See
[here](https://arrow.apache.org/docs/python/ipc.html#writing-and-reading-streams)
for more on the stream format. The following example will create an
`ArrowStreamDataset` that will connect to a host that is serving an Arrow
stream of record batches with 2 columns of dtypes=(int32, float32):

```python
import tensorflow as tf
from tensorflow_io.arrow import ArrowStreamDataset

# The str `host` should be in the format '<HOSTNAME>:<PORT>'
dataset = ArrowStreamDataset(
host,
columns=(0, 1),
output_types=(tf.int32, tf.float32),
output_shapes=([], []))

iterator = dataset.make_one_shot_iterator()
next_element = iterator.get_next()

# The host connection is made when the Dataset op is run and will iterate over
# each row of each record batch until the Arrow stream is finished
with tf.Session() as sess:
while True:
try:
print(sess.run(next_element))
except tf.errors.OutOfRangeError:
break
```

An alternate constructor can also be used to infer output types and shapes from
a given `pyarrow.Schema`, e.g. `dataset = ArrowStreamDataset.from_schema(host, schema)`
36 changes: 36 additions & 0 deletions tensorflow_io/arrow/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
# Copyright 2016 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Arrow Dataset.

@@ArrowDataset
"""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

from tensorflow_io.arrow.python.ops.arrow_dataset_ops import ArrowDataset
from tensorflow_io.arrow.python.ops.arrow_dataset_ops import ArrowFeatherDataset
from tensorflow_io.arrow.python.ops.arrow_dataset_ops import ArrowStreamDataset

from tensorflow.python.util.all_util import remove_undocumented

_allowed_symbols = [
"ArrowDataset",
"ArrowFeatherDataset",
"ArrowStreamDataset",
]

remove_undocumented(__name__)
Loading