pyspark-sql-functions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  pyspark.sql.functions module"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.master('local').appName('sparkapp2').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.abs()\n",
    "# 计算绝对值"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.acos(col)\n",
    "# 计算给定值的余弦逆;返回的角度在0到π的范围内。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(next_month=datetime.date(2015, 5, 8))]"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.add_months(start, months)\n",
    "# 返回开始后几个月的日期\n",
    "from pyspark.sql.functions import *\n",
    "df = spark.createDataFrame([('2015-04-08',)], ['dt'])\n",
    "df.select(add_months(df.dt, 1).alias('next_month')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+------+\n",
      "| name|age|height|\n",
      "+-----+---+------+\n",
      "|Alice| 10|    50|\n",
      "|  Bob| 20|    80|\n",
      "|  Tom| 30|    80|\n",
      "| Lily| 10|    30|\n",
      "+-----+---+------+\n",
      "\n",
      "+-------------+\n",
      "|distinct_afes|\n",
      "+-------------+\n",
      "|            3|\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.approxCountDistinct(col, rsd=None)\n",
    "# 同pyspark.sql.functions.approx_count_distinct(col, rsd=None)\n",
    "# 聚合函数：返回一个新的列，用于列col的近似不同计数\n",
    "# 允许的最大估计误差（默认值= 0.05）。对于rsd <0.01，使用countDistinct（）会更有效率\n",
    "df = spark.read.csv('example.csv', header=True, inferSchema=True)\n",
    "df.show()\n",
    "df.agg(approx_count_distinct(df.age).alias('distinct_afes')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|     arr|\n",
      "+--------+\n",
      "|[10, 10]|\n",
      "|[20, 20]|\n",
      "|[30, 30]|\n",
      "|[10, 10]|\n",
      "+--------+\n",
      "\n",
      "+--------+\n",
      "|     arr|\n",
      "+--------+\n",
      "|[10, 10]|\n",
      "|[20, 20]|\n",
      "|[30, 30]|\n",
      "|[10, 10]|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.array(*cols)\n",
    "# 创建一个新的数组列\n",
    "df.select(array('age', 'age').alias('arr')).show()\n",
    "df.select(array(df.age, df.age).alias('arr')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|     data|\n",
      "+---------+\n",
      "|[a, b, c]|\n",
      "|       []|\n",
      "+---------+\n",
      "\n",
      "+-----------------------+\n",
      "|array_contains(data, a)|\n",
      "+-----------------------+\n",
      "|                   true|\n",
      "|                  false|\n",
      "+-----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.array_contains(col, value)\n",
    "# 集合函数：如果数组为null，则返回null;如果数组包含给定值，则返回true;否则返回false。\n",
    "df1 = spark.createDataFrame([(['a', 'b', 'c'],), ([],)], ['data'])\n",
    "df1.show()\n",
    "df1.select(array_contains(df1.data, 'a')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+------+\n",
      "| name|age|height|\n",
      "+-----+---+------+\n",
      "|Alice| 10|    50|\n",
      "| Lily| 10|    30|\n",
      "|  Bob| 20|    80|\n",
      "|  Tom| 30|    80|\n",
      "+-----+---+------+\n",
      "\n",
      "+-----+---+------+\n",
      "| name|age|height|\n",
      "+-----+---+------+\n",
      "| Lily| 10|    30|\n",
      "|Alice| 10|    50|\n",
      "|  Bob| 20|    80|\n",
      "|  Tom| 30|    80|\n",
      "+-----+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.asc(col)\n",
    "# 基于给定列名称的升序返回一个排序表达式。\n",
    "df.sort(asc('age')).show()\n",
    "df.orderBy(df.height.asc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.ascii(col)\n",
    "# 计算字符串列的第一个字符的数字值。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.asin(col)\n",
    "# 计算给定值的正弦倒数;返回的角度在π/ 2到π/ 2的范围内。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.atan(col)\n",
    "# 计算给定值的正切倒数;返回的角度在π/ 2到π/ 2的范围内"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.base64(col)\n",
    "# 计算二进制列的BASE64编码并将其作为字符串列返回。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "|    c|\n",
      "+-----+\n",
      "| 1010|\n",
      "|10100|\n",
      "|11110|\n",
      "| 1010|\n",
      "+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.bin(col)\n",
    "# 返回给定列的二进制值的字符串表示形式。\n",
    "df.select(bin(df.age).alias('c')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.broadcast(df)\n",
    "# 将DataFrame标记为足够小以用于广播连接。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  r|\n",
      "+---+\n",
      "|2.0|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.bround(col, scale=0)\n",
    "# 如果scale> = 0，则使用HALF_EVEN舍入模式对给定值进行四舍五入以缩放小数点;如果scale <0，则将其舍入为整数部分。\n",
    "spark.createDataFrame([(2.5,)], ['a']).select(bround('a', 0).alias('r')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.cbrt(col)\n",
    "# 计算给定值的立方根。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.ceil(col)\n",
    "# 计算给定值的上限。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+\n",
      "|   a|   b|\n",
      "+----+----+\n",
      "|null|null|\n",
      "|   1|null|\n",
      "|null|   2|\n",
      "+----+----+\n",
      "\n",
      "+--------------+\n",
      "|coalesce(a, b)|\n",
      "+--------------+\n",
      "|          null|\n",
      "|             1|\n",
      "|             2|\n",
      "+--------------+\n",
      "\n",
      "+----+----+----------------+\n",
      "|   a|   b|coalesce(a, 0.0)|\n",
      "+----+----+----------------+\n",
      "|null|null|             0.0|\n",
      "|   1|null|             1.0|\n",
      "|null|   2|             0.0|\n",
      "+----+----+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.coalesce(*cols)\n",
    "# 返回非空的第一列。\n",
    "cDF = spark.createDataFrame([(None, None), (1, None), (None, 2)], ('a', 'b'))\n",
    "cDF.show()\n",
    "cDF.select(coalesce(cDF['a'], cDF['b'])).show()\n",
    "cDF.select('*', coalesce(cDF['a'], lit(0.0))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.col(col)\n",
    "# 根据给定的列名返回一个列。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(collect_list(age)=[2, 5, 5])]"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.collect_list(col)\n",
    "# 聚合函数：返回包含重复对象的列表。\n",
    "df2 = spark.createDataFrame([(2,), (5,), (5,)], ('age',))\n",
    "df2.agg(collect_list('age')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|collect_set(age)|\n",
      "+----------------+\n",
      "|          [5, 2]|\n",
      "+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.collect_set(col)\n",
    "# 聚合函数：返回一组消除重复元素的对象。\n",
    "df2.agg(collect_set('age')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "| num|\n",
      "+----+\n",
      "|1050|\n",
      "|2080|\n",
      "|3080|\n",
      "|1030|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.concat(*cols)\n",
    "# 将多个输入列连接成一个列。如果所有输入都是二进制的，则concat会以二进制形式返回输出。否则，它返回字符串。\n",
    "df.select(concat(df.age, df.height).alias('num')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|      xx|\n",
      "+--------+\n",
      "|Alice*10|\n",
      "|  Bob*20|\n",
      "|  Tom*30|\n",
      "| Lily*10|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.concat_ws(sep, *cols)\n",
    "# 使用给定的分隔符将多个输入字符串列连接到一个字符串列中。\n",
    "df.select(concat_ws('*', df.name, df.age).alias('xx')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|hex|\n",
      "+---+\n",
      "| 15|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.conv(col, fromBase, toBase)\n",
    "# 将一个字符串的数字从一个基转换成另一个基。\n",
    "df1 = spark.createDataFrame([('010101',)], ['n'])\n",
    "df1.select(conv(df1.n, 2, 16).alias('hex')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  c|\n",
      "+---+\n",
      "|1.0|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.corr(col1, col2)\n",
    "# 为col1和col2的皮尔逊相关系数返回一个新列。\n",
    "a = range(20)\n",
    "b = [2 * x for x in range(20)]\n",
    "df1 = spark.createDataFrame(zip(a, b), ['a', 'b'])\n",
    "df1.agg(corr('a', 'b').alias('c')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.cos(col)\n",
    "# 计算给定值的余弦值。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.cosh(col)\n",
    "# 计算给定值的双曲余弦。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.count(col)\n",
    "# 聚合函数：返回组中项目的数量。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  c|\n",
      "+---+\n",
      "|  4|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.countDistinct(col, *cols)\n",
    "# 为不同的col或cols返回一个新列。\n",
    "df.agg(countDistinct(df.age, df.name).alias('c')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+\n",
      "|covar_pop(a, b)|\n",
      "+---------------+\n",
      "|            0.0|\n",
      "+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.covar_pop(col1, col2)\n",
    "# 为col1和col2的人口协方差返回一个新列。\n",
    "a = [1] * 10\n",
    "b = [1] * 10\n",
    "df = spark.createDataFrame(zip(a, b), ['a', 'b'])\n",
    "df.agg(covar_pop('a', 'b')).alias('c').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  c|\n",
      "+---+\n",
      "|0.0|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.covar_samp(col1, col2)\n",
    "# 返回col1和col2的样本协方差的新列。\n",
    "df.agg(covar_samp('a', 'b').alias('c')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.crc32(col)\n",
    "# 计算二进制列的循环冗余校验值（CRC32）并将该值作为bigint返回。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(crc32=2743272264)]"
      ]
     },
     "execution_count": 74,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.createDataFrame([('ABC',)], ['a']).select(crc32('a').alias('crc32')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|          map|\n",
      "+-------------+\n",
      "|[Alice -> 10]|\n",
      "|  [Bob -> 20]|\n",
      "|  [Tom -> 30]|\n",
      "| [Lily -> 10]|\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.create_map(*cols)\n",
    "# 创建一个新的地图列。\n",
    "df.select(create_map('name', 'age').alias('map')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [],
   "source": [
    "#  pyspark.sql.functions.cume_dist()\n",
    "# 窗口函数：返回窗口分区内值的累积分布，即在当前行下面的行的分数。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.current_date()\n",
    "# 以DateType列的形式返回当前日期。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.current_timestamp()\n",
    "# 将当前时间戳作为TimestampType列返回。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "| next_date|\n",
      "+----------+\n",
      "|2015-04-09|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.date_add(start, days)\n",
    "# 返回启动后天数的日期\n",
    "df1 = spark.createDataFrame([('2015-04-08',)], ['dt'])\n",
    "df1.select(date_add(df1.dt, 1).alias('next_date')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      date|\n",
      "+----------+\n",
      "|04/08/2015|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.date_format(date, format)\n",
    "# 将日期/时间戳/字符串转换为由第二个参数给出的日期格式指定格式的字符串值。\n",
    "# 一个模式可以是例如dd.MM.yyyy，并且可以返回一个像'18 .03.1993'这样的字符串。可以使用Java类java.text.SimpleDateFormat的所有模式字母。\n",
    "# 尽可能使用像年份这样的专业功能。这些受益于专门的实施。\n",
    "df1.select(date_format('dt', 'MM/dd/yyy').alias('date')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(prev_date=datetime.date(2015, 4, 7))]"
      ]
     },
     "execution_count": 84,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.date_sub(start, days)\n",
    "# 返回开始前几天的日期\n",
    "df1.select(date_sub(df1.dt, 1).alias('prev_date')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|               year|\n",
      "+-------------------+\n",
      "|1997-01-01 00:00:00|\n",
      "+-------------------+\n",
      "\n",
      "+-------------------+\n",
      "|              month|\n",
      "+-------------------+\n",
      "|1997-02-01 00:00:00|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.date_trunc(format, timestamp)\n",
    "# 将时间戳记截断为格式指定的单位。\n",
    "df1 = spark.createDataFrame([('1997-02-28 05:02:11',)], ['t'])\n",
    "df1.select(date_trunc('year', df1.t).alias('year')).show()\n",
    "df1.select(date_trunc('mon', df1.t).alias('month')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(diff=32)]"
      ]
     },
     "execution_count": 87,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.datediff(end, start)\n",
    "# 返回从开始到结束的天数\n",
    "df1 = spark.createDataFrame([('2015-04-08','2015-05-10')], ['d1', 'd2'])\n",
    "df1.select(datediff(df1.d2, df1.d1).alias('diff')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|day|\n",
      "+---+\n",
      "|  8|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.dayofmonth(col)\n",
    "# 将给定日期的月份的日期解压为整数。\n",
    "df1 = spark.createDataFrame([('2015-04-08',)], ['dt'])\n",
    "df1.select(dayofmonth('dt').alias('day')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|day|\n",
      "+---+\n",
      "|  4|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# dayofweek 提取星期为整数\n",
    "df1.select(dayofweek('dt').alias('day')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|day|\n",
      "+---+\n",
      "| 98|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# dayofyear 提取日期为一年中第几天为整数\n",
    "df1.select(dayofyear('dt').alias('day')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.decode(col, charset）\n",
    "#使用提供的字符集（'US-ASCII'，'ISO-8859-1'，'UTF-8'，'UTF-16BE'，'UTF-16LE'之一）从二进制计算第一个参数到二进制文件中。 'UTF-16'）。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "|anInt|\n",
      "+-----+\n",
      "|    1|\n",
      "|    2|\n",
      "|    3|\n",
      "+-----+\n",
      "\n",
      "+---+-----+\n",
      "|key|value|\n",
      "+---+-----+\n",
      "|  a|    b|\n",
      "+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.explode(col)\n",
    "# 为给定数组或映射中的每个元素返回一个新行。\n",
    "from pyspark.sql import Row\n",
    "eDF = spark.createDataFrame([Row(a=1, intlist=[1,2,3], mapfield={\"a\": \"b\"})])\n",
    "eDF.select(explode(eDF.intlist).alias('anInt')).show()\n",
    "eDF.select(explode(eDF.mapfield).alias('key', 'value')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+----+-----+\n",
      "| id|  an_array| key|value|\n",
      "+---+----------+----+-----+\n",
      "|  1|[foo, bar]|   x|  1.0|\n",
      "|  2|        []|null| null|\n",
      "|  3|      null|null| null|\n",
      "+---+----------+----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.explode_outer(col)\n",
    "# 为给定数组或映射中的每个元素返回一个新行。与explode()不同，如果数组/映射为空或空，则会生成空值。\n",
    "df1 = spark.createDataFrame(\n",
    "    [(1, ['foo', 'bar'], {'x': 1.0}), (2, [], {}), (3, None, None)],\n",
    "    ('id', 'an_array', 'a_map')\n",
    ")\n",
    "df1.select('id', 'an_array', explode_outer('a_map')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+----+\n",
      "| id|     a_map| col|\n",
      "+---+----------+----+\n",
      "|  1|[x -> 1.0]| foo|\n",
      "|  1|[x -> 1.0]| bar|\n",
      "|  2|        []|null|\n",
      "|  3|      null|null|\n",
      "+---+----------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1.select('id', 'a_map', explode_outer('an_array')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+\n",
      "|length(name)|\n",
      "+------------+\n",
      "|           5|\n",
      "|           3|\n",
      "|           3|\n",
      "|           4|\n",
      "+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#  pyspark.sql.functions.expr(str)\n",
    "# 将表达式字符串分析到它表示的列中\n",
    "df.select(expr('length(name)')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  f|\n",
      "+---+\n",
      "|120|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.factorial(col)\n",
    "# 计算给定值的阶层\n",
    "df1 = spark.createDataFrame([(5,)], ['n'])\n",
    "df1.select(factorial(df1.n).alias('f')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|     v|\n",
      "+------+\n",
      "|5.0000|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.format_number(col, d)\n",
    "# 返回格式化为n位小数\n",
    "spark.createDataFrame([(5,)], ['a']).select(format_number('a', 4).alias('v')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|       v|\n",
      "+--------+\n",
      "|10 Alice|\n",
      "|  20 Bob|\n",
      "|  30 Tom|\n",
      "| 10 Lily|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#  pyspark.sql.functions.format_string(format, *cols)\n",
    "# 将参数格式化为printf-style，并将结果作为字符串列返回。\n",
    "df.select(format_string('%d %s', df.age, df.name).alias('v')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(json=Row(a=1))]"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.from_json(col, schema, options={})\n",
    "# 将包含JSON字符串的栏目解析为带有指定模式的struct类型或ArrayType的ArrayType。返回null，在不可解析字符串的情况下。\n",
    "# Spark 2.3，ddl格式的字符串或JSON格式字符串也支持schema。\n",
    "from pyspark.sql.types import *\n",
    "data = [(1, '''{\"a\":1}''')]\n",
    "schema = StructType([StructField(\"a\", IntegerType())])\n",
    "df1 = spark.createDataFrame(data, (\"key\", \"value\"))\n",
    "df1.select(from_json(df1.value, schema).alias(\"json\")).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "|json|\n",
      "+----+\n",
      "| [1]|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1.select(from_json(df1.value, 'a INT').alias('json')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "| json|\n",
      "+-----+\n",
      "|[[1]]|\n",
      "+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "schema = ArrayType(StructType([StructField('a', IntegerType())]))\n",
    "df1 = spark.createDataFrame(data, ('key', 'value'))\n",
    "df1.select(from_json(df1.value, schema).alias('json')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|                 ts|\n",
      "+-------------------+\n",
      "|2015-04-08 00:00:00|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.from_unixtime(timestamp, format='yyyy-MM-dd HH:mm:ss')\n",
    "# 将unix纪元中的秒数(1970-01-01 00:00 UTC)转换为一个字符串，该字符串表示当前系统时区中给定格式的时间戳。\n",
    "spark.conf.set('spark.sql.session.timeZone', 'America/Los_Angeles')\n",
    "time_df = spark.createDataFrame([(1428476400,)], ['unix_time'])\n",
    "time_df.select(from_unixtime('unix_time').alias('ts')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.unset(\"spark.sql.session.timeZone\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(local_time=datetime.datetime(1997, 2, 28, 2, 30))]"
      ]
     },
     "execution_count": 115,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.from_utc_timestamp(timestamp, tz)\n",
    "# 给定一个时间戳，如2017-07-14 02:40:00.0，将它解释为UTC中的时间，并将该时间呈现为给定时区中的时间戳。例如，GMT+1的收益率为2017-07-14 03:40:00.0。\n",
    "df1 = spark.createDataFrame([('1997-02-28 10:30:00',)], ['t'])\n",
    "df1.select(from_utc_timestamp(df1.t, \"PST\").alias('local_time')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+------+\n",
      "|key|     c0|    c1|\n",
      "+---+-------+------+\n",
      "|  1| value1|value2|\n",
      "|  2|value12|  null|\n",
      "+---+-------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.get_json_object(col, path)\n",
    "# 根据指定的json路径从json字符串中提取json对象，并返回提取的json对象的json字符串。如果输入json字符串无效，则返回null。\n",
    "data = [(\"1\", '''{\"f1\": \"value1\", \"f2\": \"value2\"}'''), (\"2\", '''{\"f1\": \"value12\"}''')]\n",
    "df1 = spark.createDataFrame(data, (\"key\", \"jstring\"))\n",
    "df1.select(df1.key, get_json_object(df1.jstring, '$.f1').alias(\"c0\"), \\\n",
    "           get_json_object(df1.jstring, '$.f2').alias(\"c1\") ).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|greatest|\n",
      "+--------+\n",
      "|       4|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.greatest(*cols) \n",
    "# 返回列名称列表的最大值，跳过空值。这个函数至少有两个参数。它将返回null iff所有参数都为null。\n",
    "df1 = spark.createDataFrame([(1, 4, 3)], ['a', 'b', 'c'])\n",
    "df1.select(greatest(df1.a, df1.b, df1.c).alias(\"greatest\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------+--------+\n",
      "| name|grouping(name)|sum(age)|\n",
      "+-----+--------------+--------+\n",
      "| null|             1|      70|\n",
      "|Alice|             0|      10|\n",
      "|  Bob|             0|      20|\n",
      "| Lily|             0|      10|\n",
      "|  Tom|             0|      30|\n",
      "+-----+--------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.grouping(col)\n",
    "# 聚合函数：指示GROUP BY list中指定的列是否聚合，在结果集中不聚合的聚合或0返回1。\n",
    "df.cube('name').agg(grouping('name'), sum('age')).orderBy('name').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------------+--------+\n",
      "| name|grouping_id()|sum(age)|\n",
      "+-----+-------------+--------+\n",
      "| null|            1|      70|\n",
      "|Alice|            0|      10|\n",
      "|  Bob|            0|      20|\n",
      "| Lily|            0|      10|\n",
      "|  Tom|            0|      30|\n",
      "+-----+-------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.grouping_id(*cols)\n",
    "# 列的列表应该与分组列完全匹配，或者是空的（表示所有的分组列）。\n",
    "df.cube('name').agg(grouping_id(), sum('age')).orderBy('name').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      hash|\n",
      "+----------+\n",
      "|-757602832|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.hash(*cols)\n",
    "# 计算给定列的散列码，并将结果作为int列返回。\n",
    "spark.createDataFrame([('ABC',)], ['a']).select(hash('a').alias('hash')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+\n",
      "|hex(a)|hex(b)|\n",
      "+------+------+\n",
      "|414243|     3|\n",
      "+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.hex(col)\n",
    "# 计算给定列的hex值\n",
    "spark.createDataFrame([('ABC', 3)], ['a', 'b']).select(hex('a'), hex('b')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "|hour|\n",
      "+----+\n",
      "|  13|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.hour(col) \n",
    "# 将给定日期的小时提取为整数。\n",
    "spark.createDataFrame([('2015-04-08 13:08:15',)], ['ts']).select(hour('ts').alias('hour')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.hypot(col1, col2)\n",
    "# 计算平方根"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "|    v|\n",
      "+-----+\n",
      "|Ab Cd|\n",
      "+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.initcap(col)\n",
    "# 将每个单词的第一个字母翻译成句子中的大写字母。\n",
    "spark.createDataFrame([('ab cd',)], ['a']).select(initcap(\"a\").alias('v')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|initcap(name)|\n",
      "+-------------+\n",
      "|        Alice|\n",
      "|          Bob|\n",
      "|          Tom|\n",
      "|         Lily|\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(initcap('name')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark.sql.functions.input_file_name()\n",
    "# 为当前Spark任务的文件名创建一个字符串列。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  s|\n",
      "+---+\n",
      "|  2|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.instr(str, substr) \n",
    "# 在给定的字符串中定位substr列的第一次出现的位置。如果任一参数都为null，则返回null。\n",
    "# 这个位置不是基于零的，而是基于1的索引。如果在str中不能找到substr，则返回0。\n",
    "df1 = spark.createDataFrame([('abcd',)], ['s',])\n",
    "df1.select(instr(df1.s, 'b').alias('s')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|   r1|   r2|\n",
      "+-----+-----+\n",
      "|false|false|\n",
      "| true| true|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.isnan(col)\n",
    "# 返回true iff的表达式是NaN。\n",
    "df1 = spark.createDataFrame([(1.0, float('nan')), (float('nan'), 2.0)], ('a', 'b'))\n",
    "df1.select(isnan('a').alias('r1'), isnan(df1.a).alias('r2')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|   r1|   r2|\n",
      "+-----+-----+\n",
      "|false|false|\n",
      "| true| true|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.isnull(col)\n",
    "# 返回true iff的表达式为null。\n",
    "df1 = spark.createDataFrame([(1, None), (None, 2)], ('a', 'b'))\n",
    "df1.select(isnull('a').alias('r1'), isnull(df1.a).alias('r2')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      data|\n",
      "+----------+\n",
      "|1997-02-28|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.last_day(date)\n",
    "# 返回给定日期所属的月份的最后一天。\n",
    "df1 = spark.createDataFrame([('1997-02-10',)], ['d'])\n",
    "df1.select(last_day(df1.d).alias('data')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 135,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+\n",
      "|least|\n",
      "+-----+\n",
      "|    1|\n",
      "+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.least(*cols)\n",
    "# 返回列名列表的最小值，跳过null值。这个函数至少需要两个参数。它会返回null iff所有参数都是空的。\n",
    "df1 = spark.createDataFrame([(1, 4, 3)], ['a', 'b', 'c'])\n",
    "df1.select(least(df1.a, df1.b, df1.c).alias(\"least\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 136,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|length|\n",
      "+------+\n",
      "|     4|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.length(col)\n",
    "# 计算字符串数据的字符长度或二进制数据的字节数。字符数据的长度包括后面的空格。二进制数据的长度包括二进制零。\n",
    "spark.createDataFrame([('ABC ',)], ['a']).select(length('a').alias('length')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  d|\n",
      "+---+\n",
      "|  3|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.levenshtein(left, right)\n",
    "# 计算两个给定字符串的Levenshtein距离。\n",
    "df0 = spark.createDataFrame([('kitten', 'sitting',)], ['l', 'r'])\n",
    "df0.select(levenshtein('l', 'r').alias('d')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(height=5, spark_user=True)]"
      ]
     },
     "execution_count": 138,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.lit(col)\n",
    "# 创建一个文字值的列。\n",
    "df.select(lit(5).alias('height')).withColumn('spark_user', lit(True)).take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+\n",
      "|  s|\n",
      "+---+\n",
      "|  2|\n",
      "+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.locate(substr, str, pos=1)\n",
    "# 在一个字符串列中，在位置pos之后定位substr的第一次出现的位置。\n",
    "df1 = spark.createDataFrame([('abcd',)], ['s',])\n",
    "df1.select(locate('b', df1.s, 1).alias('s')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1.0', '1.30102', '1.47712', '1.0']"
      ]
     },
     "execution_count": 142,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.log(arg1, arg2=None)\n",
    "# 返回第二个论证的第一个基于论证的对数。\n",
    "# 如果只有一个参数，那么这就取了参数的自然对数。\n",
    "df.select(log(10.0, df.age).alias('ten')).rdd.map(lambda l: str(l.ten)[:7]).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|     s|\n",
      "+------+\n",
      "|##abcd|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.lpad(col, len, pad)\n",
    "df1 = spark.createDataFrame([('abcd',)],['s',])\n",
    "df1.select(lpad(df1.s, 6, '#').alias('s')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+\n",
      "|            data|\n",
      "+----------------+\n",
      "|[1 -> a, 2 -> b]|\n",
      "+----------------+\n",
      "\n",
      "+------+\n",
      "|  keys|\n",
      "+------+\n",
      "|[1, 2]|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.map_keys(col)\n",
    "# 托收功能：返回一个无序的数组，其中包含映射的键。\n",
    "from pyspark.sql.functions import *\n",
    "df1 = spark.sql(\"SELECT map(1, 'a', 2, 'b') as data\")\n",
    "df1.show()\n",
    "df1.select(map_keys('data').alias('keys')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|values|\n",
      "+------+\n",
      "|[a, b]|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.map_values(col)\n",
    "# 托收函数：返回一个包含map值的无序数组。\n",
    "df1.select(map_values('data').alias('values')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|                hash|\n",
      "+--------------------+\n",
      "|902fbdd2b1df0c4f7...|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.md5(col)\n",
    "# 计算MD5摘要，并将值作为32个字符的hex字符串返回。\n",
    "spark.createDataFrame([('ABC',)], ['a']).select(md5('a').alias('hash')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(minute=8)]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.minute(col)\n",
    "# 将给定日期的分钟提取为整数。\n",
    "df1 = spark.createDataFrame([('2015-04-08 13:08:15',)], ['ts'])\n",
    "df1.select(minute('ts').alias('minute')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------+------------+\n",
      "|slen(name)|to_upper(name)|add_one(age)|\n",
      "+----------+--------------+------------+\n",
      "|         8|      JOHN DOE|          22|\n",
      "+----------+--------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 标量UDF中的pandas的长度不是整个输入列的长度，而是用于每次调用函数的内部批处理的长度。因此，这可以用来保证每个返回的熊猫的长度，并且不能用作列长度。\n",
    "from pyspark.sql.functions import pandas_udf, PandasUDFType\n",
    "from pyspark.sql.types import IntegerType, StringType\n",
    "slen = pandas_udf(lambda s: s.str.len(), IntegerType())\n",
    "@pandas_udf(StringType())\n",
    "def to_upper(s):\n",
    "    return s.str.upper()\n",
    "@pandas_udf('integer', PandasUDFType.SCALAR)\n",
    "def add_one(x):\n",
    "    return x + 1\n",
    "df1 = spark.createDataFrame([(1, 'John Doe', 21)],\n",
    "                           ('id', 'name', 'age'))\n",
    "df1.select(slen('name').alias('slen(name)'), to_upper('name'), add_one('age')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+\n",
      "| id|   v|\n",
      "+---+----+\n",
      "|  1| 1.0|\n",
      "|  1| 2.0|\n",
      "|  2| 3.0|\n",
      "|  2| 5.0|\n",
      "|  2|10.0|\n",
      "+---+----+\n",
      "\n",
      "+---+-------------------+\n",
      "| id|                  v|\n",
      "+---+-------------------+\n",
      "|  1|-0.7071067811865475|\n",
      "|  1| 0.7071067811865475|\n",
      "|  2|-0.8320502943378437|\n",
      "|  2|-0.2773500981126146|\n",
      "|  2| 1.1094003924504583|\n",
      "+---+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# GROUPED_MAP\n",
    "from pyspark.sql.functions import pandas_udf, PandasUDFType\n",
    "df1 = spark.createDataFrame(\n",
    "    [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],\n",
    "    ('id', 'v')\n",
    ")\n",
    "df1.show()\n",
    "@pandas_udf('id long, v double', PandasUDFType.GROUPED_MAP)\n",
    "def normalize(pdf):\n",
    "    v = pdf.v\n",
    "    return pdf.assign(v=(v - v.mean()) / v.std())\n",
    "df1.groupby('id').apply(normalize).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "用户定义的功能在默认情况下被认为是确定性的。由于优化，可能会消除重复的调用，甚至可能会调用该函数的次数超过查询中的次数。如果您的功能不确定，请在用户定义的功能上调用asNondeterministic。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "@pandas_udf('double', PandasUDFType.SCALAR)\n",
    "def random(v):\n",
    "    import numpy as np\n",
    "    import pandas as pd\n",
    "    return pd.Series(np.random.randn(len(v)))\n",
    "random = random.asNondeterministic()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+\n",
      "|pos|col|\n",
      "+---+---+\n",
      "|  0|  1|\n",
      "|  1|  2|\n",
      "|  2|  3|\n",
      "+---+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.posexplode(col)\n",
    "# 为给定数组或地图中的每个元素返回一个新行。\n",
    "from pyspark.sql import Row\n",
    "eDF = spark.createDataFrame([Row(a=1, intlist=[1, 2, 3], mapfield={'a':'b'})])\n",
    "eDF.select(posexplode(eDF.intlist)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+-----+\n",
      "|pos|key|value|\n",
      "+---+---+-----+\n",
      "|  0|  a|    b|\n",
      "+---+---+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "eDF.select(posexplode(eDF.mapfield)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+----+----+-----+\n",
      "| id|  an_array| pos| key|value|\n",
      "+---+----------+----+----+-----+\n",
      "|  1|[foo, bar]|   0|   x|  1.0|\n",
      "|  2|        []|null|null| null|\n",
      "|  3|      null|null|null| null|\n",
      "+---+----------+----+----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.posexplode_outer(col)\n",
    "# 为给定数组或地图中的每个元素返回一个新行。与posexplode不同的是，如果数组/映射为null或为空，则会生成该行（null，null）。\n",
    "df1 = spark.createDataFrame(\n",
    "    [(1, [\"foo\", \"bar\"], {\"x\": 1.0}), (2, [], {}), (3, None, None)],\n",
    "    (\"id\", \"an_array\", \"a_map\")\n",
    ")\n",
    "df1.select('id', 'an_array', posexplode_outer('a_map')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------+----+----+\n",
      "| id|     a_map| pos| col|\n",
      "+---+----------+----+----+\n",
      "|  1|[x -> 1.0]|   0| foo|\n",
      "|  1|[x -> 1.0]|   1| bar|\n",
      "|  2|        []|null|null|\n",
      "|  3|      null|null|null|\n",
      "+---+----------+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df1.select('id', 'a_map', posexplode_outer('an_array')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+------+------------------+\n",
      "| name|age|height|              rand|\n",
      "+-----+---+------+------------------+\n",
      "|Alice| 10|    50|1.9983710323241177|\n",
      "|  Bob| 20|    80|2.5749454053758716|\n",
      "|  Tom| 30|    80|2.7419891047485545|\n",
      "| Lily| 10|    30|2.5994827668473834|\n",
      "+-----+---+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.rand(seed=None)\n",
    "# 从U [0.0，1.0]生成一个具有独立且分布相同（i.i.d.）样本的随机列。\n",
    "df.withColumn('rand', rand(seed=42) * 3).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+---+------+------------------+\n",
      "| name|age|height|             randn|\n",
      "+-----+---+------+------------------+\n",
      "|Alice| 10|    50|0.4085363219031828|\n",
      "|  Bob| 20|    80|0.8811793095417685|\n",
      "|  Tom| 30|    80|-2.013921870967947|\n",
      "| Lily| 10|    30|1.6641751435679302|\n",
      "+-----+---+------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.randn(seed=None)\n",
    "# 从标准正态分布生成具有独立且分布相同（i.i.d.）样本的列。\n",
    "df.withColumn('randn', randn(42)).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(d='100')]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.regexp_extract(str, pattern, idx)\n",
    "# 从指定的字符串列中提取由Java正则表达式匹配的特定组。如果正则表达式不匹配，或者指定的组不匹配，则返回空字符串。\n",
    "df1 = spark.createDataFrame([('100-200',)], ['str'])\n",
    "df1.select(regexp_extract('str', '(\\d+)-(\\d+)', 1).alias('d')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(d='oo')]"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df1 = spark.createDataFrame([('foo',)], ['str'])\n",
    "df1.select(regexp_extract('str', '(o.*)', 1).alias('d')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(d='xx-xx')]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.regexp_replace(str, pattern, replacement)\n",
    "# 将与regexp匹配的指定字符串值的所有子字符串替换为rep。\n",
    "df1 = spark.createDataFrame([('100-200',)], ['str'])\n",
    "df1.select(regexp_replace('str', '(\\d+)', 'xx').alias('d')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|     s|\n",
      "+------+\n",
      "|ababab|\n",
      "+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.repeat(col, n)\n",
    "# 重复一次字符串列n次，并将其作为新的字符串列返回。\n",
    "df1 = spark.createDataFrame([('ab',)], ['s',])\n",
    "df1.select(repeat(df1.s, 3).alias('s')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(r=[1, 2, 3]), Row(r=[1]), Row(r=[])]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.sort_array(col, asc=True)\n",
    "# 汇集函数：根据数组元素的自然顺序，按照升序或降序对输入数组进行排序。\n",
    "df1 = spark.createDataFrame([([2, 1, 3],),([1],),([],)], ['data'])\n",
    "df1.select(sort_array(df1.data).alias('r')).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(soundex='P362'), Row(soundex='U612')]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# pyspark.sql.functions.soundex(col）\n",
    "# 返回字符串的SoundEx编码\n",
    "df1 = spark.createDataFrame([(\"Peters\",),(\"Uhrbach\",)], ['name'])\n",
    "df1.select(soundex(df1.name).alias(\"soundex\")).collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|       s|\n",
      "+--------+\n",
      "|[ab, cd]|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# pyspark.sql.functions.split(str, pattern)\n",
    "# 将模式分割（pattern是一个正则表达式）。\n",
    "df1 = spark.createDataFrame([('ab12cd',)], ['s',])\n",
    "df1.select(split(df1.s, '[0-9]+').alias('s')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}