notebook中链接Hadoop生态 - tencentmusic/cube-studio GitHub Wiki

import findspark
findspark.init()

os.environ['SPARK_HOME'] = "/opt/third/spark"
sys.path.append("/opt/third/spark/python")
sys.path.append("/opt/third/spark/python/lib/py4j-0.10.9-src.zip")

import os
from random import random
from operator import add

import findspark

os.environ['SPARK_HOME']='/opt/third/spark'
findspark.init()

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession\
    .builder\
    .appName("PythonPi-Local")\
    .master("local")\
    .getOrCreate()

    n = 100000 * 2

    def f(_):
        x = random() * 2 - 1
        y = random() * 2 - 1
        return 1 if x ** 2 + y ** 2 <= 1 else 0

    count = spark.sparkContext.parallelize(range(1, n + 1), 2).map(f).reduce(add)
    print("Pi is roughly %f" % (4.0 * count / n))

    spark.stop()

import os
import findspark

os.environ['SPARK_HOME']='/opt/third/spark'
findspark.init()

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession.builder \
    .appName('spark-hive-demo') \
    .config("hive.metastore.uris", "thrift://xxx.xxx.xxx.xxx:9083") \
    .enableHiveSupport() \
    .getOrCreate()

    spark.sql("create table if not exists demo(id bigint,name String)")

    spark.sql("insert overwrite demo values (1,'hamawhite'),(2,'song.bs')")
    spark.sql("select * from demo").show()

import os
import sys
from random import random
from operator import add

import findspark

os.environ['SPARK_HOME']='/opt/third/spark'
os.environ['HADOOP_CONF_DIR']='/opt/third/hadoop/etc/hadoop'

findspark.init()

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession\
    .builder\
    .appName("PythonPi-Yarn-Client")\
    .master("yarn")\
    .config("spark.submit.deployMode", "client")\
    .getOrCreate()

    n = 100000 * 2

    def f(_):
        x = random() * 2 - 1
        y = random() * 2 - 1
        return 1 if x ** 2 + y ** 2 <= 1 else 0

    count = spark.sparkContext.parallelize(range(1, n + 1), 2).map(f).reduce(add)
    print("Pi is roughly %f" % (4.0 * count / n))

    spark.stop()

# 删除已有的JDK11
rm -rf /usr/lib/jvm/

# 安装JDK8
apt-get update
apt-get install -y openjdk-8-jdk

# 设置下环境变量
vi ~/.bashrc

#java
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export PATH=$PATH:$JAVA_HOME/bin

source ~/.bashrc

# 测试下java命令
java -version

from pyflink.table import EnvironmentSettings, TableEnvironment

env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

t_env.execute_sql("""
    CREATE TABLE random_source(
        id BIGINT, 
        data TINYINT
    ) WITH (
        'connector' = 'datagen',
        'fields.id.kind'='sequence',
        'fields.id.start'='1',
        'fields.id.end'='8',
        'fields.data.kind'='sequence',
        'fields.data.start'='4',
        'fields.data.end'='11'
    )
""")

t_env.execute_sql("""
    CREATE TABLE print_sink (
        id BIGINT, 
        data_sum TINYINT 
    ) WITH (
        'connector' = 'print'
    )
""")

t_env.execute_sql("""
    INSERT INTO print_sink 
        SELECT id, sum(data) as data_sum FROM 
            (SELECT id / 2 as id, data FROM random_source )
        WHERE id > 1
        GROUP BY id
""").wait()

import os 
from pyflink.table import EnvironmentSettings, TableEnvironment

env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

flink_lib_path="/opt/third/flink/lib"
jars = []
for file in os.listdir(flink_lib_path):
    if file.endswith('.jar'):
        jars.append(os.path.basename(file))
str_jars = ';'.join(['file://'+flink_lib_path +'/'+ jar for jar in jars])
t_env.get_config().get_configuration().set_string("pipeline.jars", str_jars)

from pyflink.table.catalog import HiveCatalog

# Create a HiveCatalog
catalog_name = "hive"
default_database = "default"
catalog = HiveCatalog(catalog_name, default_database, "/opt/third/hive/conf")
t_env.register_catalog(catalog_name, catalog)
t_env.use_catalog(catalog_name)

t_env.execute_sql("DROP TABLE IF EXISTS random_source_pyflink")
t_env.execute_sql("""
    CREATE TABLE IF NOT EXISTS random_source_pyflink (
        id BIGINT, 
        data TINYINT
    ) WITH (
        'connector' = 'datagen',
        'fields.id.kind'='sequence',
        'fields.id.start'='1',
        'fields.id.end'='8',
        'fields.data.kind'='sequence',
        'fields.data.start'='4',
        'fields.data.end'='11'
    )
""")

t_env.execute_sql("DROP TABLE IF EXISTS print_sink_pyflink")
t_env.execute_sql("""
    CREATE TABLE IF NOT EXISTS print_sink_pyflink  (
        id BIGINT, 
        data_sum TINYINT 
    ) WITH (
        'connector' = 'print'
    )
""")

t_env.execute_sql("""
    INSERT INTO print_sink_pyflink 
        SELECT id, sum(data) as data_sum FROM 
            (SELECT id / 2 as id, data FROM random_source_pyflink )
        WHERE id > 1
        GROUP BY id
""").wait()

import os
import findspark

os.environ['SPARK_HOME']='/opt/third/spark'
findspark.init()

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession.builder \
    .appName('spark-hive-demo') \
    .config("hive.metastore.uris", "thrift://192.168.90.150:9083") \
    .enableHiveSupport() \
    .getOrCreate()

    spark.sql("show tables").show()

from random import random
from operator import add
from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession\
    .builder\
    .appName("PythonPi-Local")\
    .master("local")\
    .getOrCreate()

    n = 100000 * 2

    def f(_):
        x = random() * 2 - 1
        y = random() * 2 - 1
        return 1 if x ** 2 + y ** 2 <= 1 else 0

    count = spark.sparkContext.parallelize(range(1, n + 1), 2).map(f).reduce(add)
    print("Pi is roughly %f" % (4.0 * count / n))

    spark.stop()

from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession.builder \
    .appName('spark-hive-demo') \
    .config("hive.metastore.uris", "thrift://xxx.xxx.xxx.xxx:9083") \
    .enableHiveSupport() \
    .getOrCreate()

    spark.sql("create table if not exists demo(id bigint,name String)")

    spark.sql("insert overwrite demo values (1,'hamawhite'),(2,'song.bs')")
    spark.sql("select * from demo").show()

from random import random
from operator import add
from pyspark.sql import SparkSession

if __name__ == "__main__":
    spark = SparkSession\
    .builder\
    .appName("PythonPi-Yarn-Client-Dockerfile")\
    .master("yarn")\
    .config("spark.submit.deployMode", "client")\
    .getOrCreate()

    n = 100000 * 2

    def f(_):
        x = random() * 2 - 1
        y = random() * 2 - 1
        return 1 if x ** 2 + y ** 2 <= 1 else 0

    count = spark.sparkContext.parallelize(range(1, n + 1), 2).map(f).reduce(add)
    print("Pi is roughly %f" % (4.0 * count / n))

    spark.stop()