| ##### get current directory in absolute path | |
| # Add `> /dev/null 2>&1` to eliminate outputs of `cd` in some cases | |
| base_dir="$( cd "$(dirname $0)" > /dev/null 2>&1 && pwd)" | |
| # other options | |
| current_dir=$(pwd) # current working directory | |
| base_dir=$(dirname $0) # in relative path |
| ESRunner runner = new ESRunner(); | |
| // Gracefully shutdown | |
| final Thread mainThread = Thread.currentThread(); | |
| Runtime.getRuntime().addShutdownHook(new Thread(() -> { | |
| try { | |
| runner.shutdown(); // Clean up work | |
| mainThread.join(200); // wait main thread to stop |
| """ | |
| Turn this | |
| days name | |
| [1,3,5,7] John | |
| into this | |
| days name | |
| 1 John |
| """ | |
| Turn this | |
| location name Jan-2010 Feb-2010 March-2010 | |
| A "test" 12 20 30 | |
| B "foo" 18 20 25 | |
| into this | |
| location name Date Value |
UDF can return only a single column at the time. There are two different ways you can overcome this limitation:
The most general solution is a StructType but you can consider ArrayType or MapType as well.
import org.apache.spark.sql.functions.udf
val df = Seq(
(1L, 3.0, "a"), (2L, -1.0, "b"), (3L, 0.0, "c")
Suppose you need to apply the same function to multiple columns in one DataFrame, one straight way is like this:
val newDF = oldDF.withColumn("colA", func("colA")).withColumn("colB", func("colB")).withColumn("colC", func("colC"))If you want to save some type, you can try this:
select with varargs including *:import spark.implicits._
curl http://www.centos.org
| # 附带一个用spark将数据取回本地用于绘图的方法 | |
| def toArr(df, col, dtype=np.int32): | |
| """ | |
| 将DataFrame的一列取回本地，并转成numpy.ndarray格式。 | |
| df: 目标DataFrame | |
| col: 目标列名 | |
| dtype: 目标列的数据格式 | |
| return: 目标列的数据。np.ndarray | |
| """ |
| %matplotlib inline | |
| buckets = [-87.0, -15, 0, 30, 120] | |
| rdd_histogram_data = ml_bucketized_features\ | |
| .select("ArrDelay")\ | |
| .rdd\ | |
| .flatMap(lambda x: x)\ | |
| .histogram(buckets) | |
| create_hist(rdd_histogram_data) |