milenkovicm commented on code in PR #1100:
URL:
https://github.com/apache/datafusion-ballista/pull/1100#discussion_r1844661189
##########
docs/source/user-guide/python.md:
##########
@@ -103,14 +119,15 @@ The `explain` method can be used to show the logical and
physical query plans fo
The following example demonstrates creating arrays with PyArrow and then
creating a Ballista DataFrame.
```python
-import ballista
+from ballista import Ballista, BallistaBuilder
Review Comment:
Why do we need `Ballista`, looks like everything can be covered with
`BallistaBuilder`
##########
python/src/lib.rs:
##########
@@ -15,18 +15,189 @@
// specific language governing permissions and limitations
// under the License.
+use ballista::prelude::*;
+use datafusion_python::context::PySessionContext as
DataFusionPythonSessionContext;
+use datafusion_python::utils::wait_for_future;
+
+use std::collections::HashMap;
+
use pyo3::prelude::*;
-pub mod context;
mod utils;
-
-pub use crate::context::PySessionContext;
+use utils::to_pyerr;
#[pymodule]
-fn pyballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
+fn ballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
pyo3_log::init();
// Ballista structs
- m.add_class::<PySessionContext>()?;
+ m.add_class::<PyBallista>()?;
+ m.add_class::<PyBallistaBuilder>()?;
// DataFusion structs
m.add_class::<datafusion_python::dataframe::PyDataFrame>()?;
+ // Ballista Config
+ /*
+ // Future implementation will include more state and config options
+ m.add_class::<PySessionStateBuilder>()?;
+ m.add_class::<PySessionState>()?;
+ m.add_class::<PySessionConfig>()?;
+ */
Ok(())
}
+
+// Ballista Builder will take a HasMap/Dict Cionfg
+#[pyclass(name = "BallistaBuilder", module = "ballista", subclass)]
+pub struct PyBallistaBuilder {
+ conf: HashMap<String, String>,
+}
+
+#[pymethods]
+impl PyBallistaBuilder {
+ #[new]
+ pub fn new() -> Self {
+ Self { conf: HashMap::new() }
+ }
+
+ pub fn set(
+ mut slf: PyRefMut<'_, Self>,
+ k: &str,
+ v: &str,
+ py: Python,
+ ) -> PyResult<PyObject> {
+ slf.conf.insert(k.into(), v.into());
+
+ Ok(slf.into_py(py))
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.iter() {
+ println!("\t{}: {}", ele.0, ele.1)
+ }
+ }
+
+ pub fn build(slf: PyRef<'_, Self>) -> PyBallista {
+ PyBallista {
+ conf: PyBallistaBuilder {
+ conf: slf.conf.clone(),
+ },
+ }
+ }
+}
+
+#[pyclass(name = "Ballista", module = "ballista", subclass)]
+pub struct PyBallista {
+ pub conf: PyBallistaBuilder,
+}
+
+#[pymethods]
+impl PyBallista {
+ #[new]
+ pub fn new() -> Self {
+ Self {
+ conf: PyBallistaBuilder::new(),
+ }
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.conf.clone() {
+ println!("{:4}: {}", ele.0, ele.1)
+ }
+ }
+
+ /// Construct the standalone instance from the SessionContext
+ #[pyo3(signature = (concurrent_tasks = 4))]
+ pub fn standalone(
+ &self,
+ concurrent_tasks: usize,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Define the SessionContext
+ let session_context = BallistaContext::standalone(&config,
concurrent_tasks);
+ // SessionContext is an async function
+ let ctx = wait_for_future(py, session_context)
+ .map_err(to_pyerr)?
+ .context()
+ .clone();
+
+ // Convert the SessionContext into a Python SessionContext
+ Ok(ctx.into())
+ }
+
+ /// Construct the remote instance from the SessionContext
+ pub fn remote(
+ &self,
+ host: &str,
+ port: u16,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Create the BallistaContext
+ let session_context = BallistaContext::remote(host, port, config);
+ let ctx = wait_for_future(py, session_context)
+ .map_err(to_pyerr)?
+ .context()
+ .clone();
+
+ // Convert the SessionContext into a Python SessionContext
+ Ok(ctx.into())
+ }
+}
+
+/*
+Plan to implement Session Config and State in a future issue
Review Comment:
I don't see reason for `PySessionConfig` why do you think it is important?
Can we get it from datafusion-python
##########
python/ballista/context.py:
##########
@@ -0,0 +1,79 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+from _typeshed import Self
+from datafusion import SessionContext
+from ballista import Ballista, BallistaBuilder
+
+from typing import List, Any
+
+
+class BallistaBuilder:
+ def __new__(cls):
+ return super().__new__(cls)
+
+ def __init__(self) -> None:
+ self = {}
+
+ def set(self, k, v) -> Self:
+ return self.set(k, v)
+
+
+class Ballista:
+ def __new__(cls):
+ return super().__new__(cls)
+
+ def __init__(self) -> None:
+ self.state = Ballista()
+
+ def standalone(self, concurrent_tasks: int = 4):
Review Comment:
`concurrent_task` is not needed as we have configuration property for that
now
##########
python/examples/example.py:
##########
@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+from ballista import Ballista, BallistaBuilder
Review Comment:
I would suggest removing `Ballista`, IMHO, makes no sense to import two
classes to bootstrap Context
##########
python/src/lib.rs:
##########
@@ -15,18 +15,189 @@
// specific language governing permissions and limitations
// under the License.
+use ballista::prelude::*;
+use datafusion_python::context::PySessionContext as
DataFusionPythonSessionContext;
+use datafusion_python::utils::wait_for_future;
+
+use std::collections::HashMap;
+
use pyo3::prelude::*;
-pub mod context;
mod utils;
-
-pub use crate::context::PySessionContext;
+use utils::to_pyerr;
#[pymodule]
-fn pyballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
+fn ballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
pyo3_log::init();
// Ballista structs
- m.add_class::<PySessionContext>()?;
+ m.add_class::<PyBallista>()?;
+ m.add_class::<PyBallistaBuilder>()?;
// DataFusion structs
m.add_class::<datafusion_python::dataframe::PyDataFrame>()?;
+ // Ballista Config
+ /*
+ // Future implementation will include more state and config options
+ m.add_class::<PySessionStateBuilder>()?;
+ m.add_class::<PySessionState>()?;
+ m.add_class::<PySessionConfig>()?;
+ */
Ok(())
}
+
+// Ballista Builder will take a HasMap/Dict Cionfg
+#[pyclass(name = "BallistaBuilder", module = "ballista", subclass)]
+pub struct PyBallistaBuilder {
+ conf: HashMap<String, String>,
+}
+
+#[pymethods]
+impl PyBallistaBuilder {
+ #[new]
+ pub fn new() -> Self {
+ Self { conf: HashMap::new() }
+ }
+
+ pub fn set(
+ mut slf: PyRefMut<'_, Self>,
+ k: &str,
+ v: &str,
+ py: Python,
+ ) -> PyResult<PyObject> {
+ slf.conf.insert(k.into(), v.into());
+
+ Ok(slf.into_py(py))
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.iter() {
+ println!("\t{}: {}", ele.0, ele.1)
+ }
+ }
+
+ pub fn build(slf: PyRef<'_, Self>) -> PyBallista {
+ PyBallista {
+ conf: PyBallistaBuilder {
+ conf: slf.conf.clone(),
+ },
+ }
+ }
+}
+
+#[pyclass(name = "Ballista", module = "ballista", subclass)]
+pub struct PyBallista {
+ pub conf: PyBallistaBuilder,
+}
+
+#[pymethods]
+impl PyBallista {
+ #[new]
+ pub fn new() -> Self {
+ Self {
+ conf: PyBallistaBuilder::new(),
+ }
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.conf.clone() {
+ println!("{:4}: {}", ele.0, ele.1)
+ }
+ }
+
+ /// Construct the standalone instance from the SessionContext
+ #[pyo3(signature = (concurrent_tasks = 4))]
+ pub fn standalone(
+ &self,
+ concurrent_tasks: usize,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Define the SessionContext
+ let session_context = BallistaContext::standalone(&config,
concurrent_tasks);
Review Comment:
please dont use `BallistaContext` use
```rust
let session_config = SessionConfig::new_with_ballista()
.with_information_schema(true)
.with_ballista_job_name("Super Cool Ballista App");
let state = SessionStateBuilder::new()
.with_default_features()
.with_config(session_config)
.build();
let ctx: SessionContext =
SessionContext::standalone_with_state(state).await?;
```
##########
python/examples/example.py:
##########
@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+from ballista import Ballista, BallistaBuilder
+from datafusion.context import SessionContext
+
+# Ballista will initiate with an empty config
+# set config variables with `set()`
+ballista = BallistaBuilder()\
+ .set("ballista.job.name", "example ballista")\
+ .set("ballista.shuffle.partitions", "16")\
+ .set("ballista.executor.cpus", "4")\
+ .build()
+
+# Show the Ballista Config
+print(ballista.show_config())
+
+# Build a standalone Cluster (use `remote()`)
+# for remote cluster
+ctx: SessionContext = ballista.standalone()
Review Comment:
I would suggest change
```python
ctx: SessionContext = BallistaBuilder()
.config("ballista.job.name", "example ballista")
.config("datafusion.execution.target_partitions", "16")
.standalone()
```
##########
python/ballista/context.py:
##########
@@ -0,0 +1,79 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+from _typeshed import Self
+from datafusion import SessionContext
+from ballista import Ballista, BallistaBuilder
+
+from typing import List, Any
+
+
+class BallistaBuilder:
+ def __new__(cls):
+ return super().__new__(cls)
+
+ def __init__(self) -> None:
+ self = {}
+
+ def set(self, k, v) -> Self:
+ return self.set(k, v)
+
+
+class Ballista:
+ def __new__(cls):
+ return super().__new__(cls)
+
+ def __init__(self) -> None:
+ self.state = Ballista()
+
+ def standalone(self, concurrent_tasks: int = 4):
+ self.standalone()
+
+ def builder(self) -> BallistaBuilder:
+ return BallistaBuilder()
+
+"""
Review Comment:
do we need to document this?
##########
python/src/lib.rs:
##########
@@ -15,18 +15,189 @@
// specific language governing permissions and limitations
// under the License.
+use ballista::prelude::*;
+use datafusion_python::context::PySessionContext as
DataFusionPythonSessionContext;
+use datafusion_python::utils::wait_for_future;
+
+use std::collections::HashMap;
+
use pyo3::prelude::*;
-pub mod context;
mod utils;
-
-pub use crate::context::PySessionContext;
+use utils::to_pyerr;
#[pymodule]
-fn pyballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
+fn ballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
pyo3_log::init();
// Ballista structs
- m.add_class::<PySessionContext>()?;
+ m.add_class::<PyBallista>()?;
+ m.add_class::<PyBallistaBuilder>()?;
// DataFusion structs
m.add_class::<datafusion_python::dataframe::PyDataFrame>()?;
+ // Ballista Config
+ /*
+ // Future implementation will include more state and config options
+ m.add_class::<PySessionStateBuilder>()?;
+ m.add_class::<PySessionState>()?;
+ m.add_class::<PySessionConfig>()?;
+ */
Ok(())
}
+
+// Ballista Builder will take a HasMap/Dict Cionfg
+#[pyclass(name = "BallistaBuilder", module = "ballista", subclass)]
+pub struct PyBallistaBuilder {
+ conf: HashMap<String, String>,
+}
+
+#[pymethods]
+impl PyBallistaBuilder {
+ #[new]
+ pub fn new() -> Self {
+ Self { conf: HashMap::new() }
+ }
+
+ pub fn set(
+ mut slf: PyRefMut<'_, Self>,
+ k: &str,
+ v: &str,
+ py: Python,
+ ) -> PyResult<PyObject> {
+ slf.conf.insert(k.into(), v.into());
+
+ Ok(slf.into_py(py))
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.iter() {
+ println!("\t{}: {}", ele.0, ele.1)
+ }
+ }
+
+ pub fn build(slf: PyRef<'_, Self>) -> PyBallista {
+ PyBallista {
+ conf: PyBallistaBuilder {
+ conf: slf.conf.clone(),
+ },
+ }
+ }
+}
+
+#[pyclass(name = "Ballista", module = "ballista", subclass)]
+pub struct PyBallista {
+ pub conf: PyBallistaBuilder,
+}
+
+#[pymethods]
+impl PyBallista {
+ #[new]
+ pub fn new() -> Self {
+ Self {
+ conf: PyBallistaBuilder::new(),
+ }
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.conf.clone() {
+ println!("{:4}: {}", ele.0, ele.1)
+ }
+ }
+
+ /// Construct the standalone instance from the SessionContext
+ #[pyo3(signature = (concurrent_tasks = 4))]
+ pub fn standalone(
+ &self,
+ concurrent_tasks: usize,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Define the SessionContext
+ let session_context = BallistaContext::standalone(&config,
concurrent_tasks);
+ // SessionContext is an async function
+ let ctx = wait_for_future(py, session_context)
+ .map_err(to_pyerr)?
+ .context()
+ .clone();
+
+ // Convert the SessionContext into a Python SessionContext
+ Ok(ctx.into())
+ }
+
+ /// Construct the remote instance from the SessionContext
+ pub fn remote(
+ &self,
+ host: &str,
+ port: u16,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Create the BallistaContext
+ let session_context = BallistaContext::remote(host, port, config);
Review Comment:
please dont use `BallistaContext` use
```rust
let session_config = SessionConfig::new_with_ballista()
.with_information_schema(true)
.with_ballista_job_name("Super Cool Ballista App");
let state = SessionStateBuilder::new()
.with_default_features()
.with_config(session_config)
.build();
let ctx: SessionContext = SessionContext::remote_with_state(&url,
state).await?;
```
##########
docs/source/user-guide/python.md:
##########
@@ -28,9 +28,25 @@ popular file formats files, run it in a distributed
environment, and obtain the
The following code demonstrates how to create a Ballista context and connect
to a scheduler.
+If you are running a standalone cluster (runs locally), all you need to do is
call the stand alone cluster method `standalone()` or your BallistaContext. If
you are running a cluster in remote mode, you need to provide the URL
`Ballista.remote("http://my-remote-ip:50050";)`.
+
```text
->>> import ballista
->>> ctx = ballista.BallistaContext("localhost", 50050)
+>>> from ballista import Ballista, BallistaBuilder
+>>> # for a standalone instance
+>>> # Ballista will initiate with an empty config
+>>> # set config variables with `set()`
+>>> ballista = BallistaBuilder()\
+>>> .set("ballista.job.name", "example ballista")\
+>>> .set("ballista.shuffle.partitions", "16")\
+>>> .set("ballista.executor.cpus", "4")\
+>>> .build()
Review Comment:
i think it would make more sense if at this point we call `standalone` or
`remote` to get the context. I'm not sure what's the benefit of
`.build().stadndalone()`
##########
python/src/lib.rs:
##########
@@ -15,18 +15,189 @@
// specific language governing permissions and limitations
// under the License.
+use ballista::prelude::*;
+use datafusion_python::context::PySessionContext as
DataFusionPythonSessionContext;
+use datafusion_python::utils::wait_for_future;
+
+use std::collections::HashMap;
+
use pyo3::prelude::*;
-pub mod context;
mod utils;
-
-pub use crate::context::PySessionContext;
+use utils::to_pyerr;
#[pymodule]
-fn pyballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
+fn ballista_internal(_py: Python, m: Bound<'_, PyModule>) -> PyResult<()> {
pyo3_log::init();
// Ballista structs
- m.add_class::<PySessionContext>()?;
+ m.add_class::<PyBallista>()?;
+ m.add_class::<PyBallistaBuilder>()?;
// DataFusion structs
m.add_class::<datafusion_python::dataframe::PyDataFrame>()?;
+ // Ballista Config
+ /*
+ // Future implementation will include more state and config options
+ m.add_class::<PySessionStateBuilder>()?;
+ m.add_class::<PySessionState>()?;
+ m.add_class::<PySessionConfig>()?;
+ */
Ok(())
}
+
+// Ballista Builder will take a HasMap/Dict Cionfg
+#[pyclass(name = "BallistaBuilder", module = "ballista", subclass)]
+pub struct PyBallistaBuilder {
+ conf: HashMap<String, String>,
+}
+
+#[pymethods]
+impl PyBallistaBuilder {
+ #[new]
+ pub fn new() -> Self {
+ Self { conf: HashMap::new() }
+ }
+
+ pub fn set(
+ mut slf: PyRefMut<'_, Self>,
+ k: &str,
+ v: &str,
+ py: Python,
+ ) -> PyResult<PyObject> {
+ slf.conf.insert(k.into(), v.into());
+
+ Ok(slf.into_py(py))
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.iter() {
+ println!("\t{}: {}", ele.0, ele.1)
+ }
+ }
+
+ pub fn build(slf: PyRef<'_, Self>) -> PyBallista {
+ PyBallista {
+ conf: PyBallistaBuilder {
+ conf: slf.conf.clone(),
+ },
+ }
+ }
+}
+
+#[pyclass(name = "Ballista", module = "ballista", subclass)]
+pub struct PyBallista {
+ pub conf: PyBallistaBuilder,
+}
+
+#[pymethods]
+impl PyBallista {
+ #[new]
+ pub fn new() -> Self {
+ Self {
+ conf: PyBallistaBuilder::new(),
+ }
+ }
+
+ pub fn show_config(&self) {
+ println!("Ballista Config:");
+ for ele in self.conf.conf.clone() {
+ println!("{:4}: {}", ele.0, ele.1)
+ }
+ }
+
+ /// Construct the standalone instance from the SessionContext
+ #[pyo3(signature = (concurrent_tasks = 4))]
+ pub fn standalone(
+ &self,
+ concurrent_tasks: usize,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Define the SessionContext
+ let session_context = BallistaContext::standalone(&config,
concurrent_tasks);
+ // SessionContext is an async function
+ let ctx = wait_for_future(py, session_context)
+ .map_err(to_pyerr)?
+ .context()
+ .clone();
+
+ // Convert the SessionContext into a Python SessionContext
+ Ok(ctx.into())
+ }
+
+ /// Construct the remote instance from the SessionContext
+ pub fn remote(
+ &self,
+ host: &str,
+ port: u16,
+ py: Python,
+ ) -> PyResult<DataFusionPythonSessionContext> {
+ // Build the config
+ let config =
&BallistaConfig::with_settings(self.conf.conf.clone()).unwrap();
+ // Create the BallistaContext
+ let session_context = BallistaContext::remote(host, port, config);
+ let ctx = wait_for_future(py, session_context)
+ .map_err(to_pyerr)?
+ .context()
+ .clone();
+
+ // Convert the SessionContext into a Python SessionContext
+ Ok(ctx.into())
+ }
+}
+
+/*
+Plan to implement Session Config and State in a future issue
Review Comment:
Also, please remove comment if it is dead code
--
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: github-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
