programing

데이터베이스 테이블을 스파크 데이터 프레임으로 읽기 위해 Apache Spark와 MySQL을 통합하는 방법은 무엇입니까?

goodsources 2023. 10. 13. 22:06
반응형

데이터베이스 테이블을 스파크 데이터 프레임으로 읽기 위해 Apache Spark와 MySQL을 통합하는 방법은 무엇입니까?

Apache Spark와 MySQL로 기존 애플리케이션을 실행하고 싶습니다.

pySpark를 통해 저에게 도움이 됩니다.

dataframe_mysql = mySqlContext.read.format("jdbc").options(
    url="jdbc:mysql://localhost:3306/my_bd_name",
    driver = "com.mysql.jdbc.Driver",
    dbtable = "my_tablename",
    user="root",
    password="root").load()

스파크 2.0.x를 사용하면 DataFrameReader와 DataFrameWriter를 사용할 수 있습니다.SparkSession.read를 사용하여 DataFrameReader에 액세스하고 Dataset을 사용합니다.DataFrameWriter에 액세스하기 위해 쓰기를 수행합니다.

스파크 셸을 사용한다고 가정합니다.

예문을 읽다

val prop=new java.util.Properties()
prop.put("user","username")
prop.put("password","yourpassword")
val url="jdbc:mysql://host:port/db_name"

val df=spark.read.jdbc(url,"table_name",prop) 
df.show()

예문을 읽다 2

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load()

스파크 문서에서

보기3

테이블이 아닌 쿼리 결과에서 데이터를 읽으려는 경우.

val sql="""select * from db.your_table where id>1"""
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql:dbserver")
  .option("dbtable",  s"( $sql ) t")
  .option("user", "username")
  .option("password", "password")
  .load()

예문을 쓰다

import org.apache.spark.sql.SaveMode

val prop=new java.util.Properties()
prop.put("user","username")
prop.put("password","yourpassword")
val url="jdbc:mysql://host:port/db_name"
//df is a dataframe contains the data which you want to write.
df.write.mode(SaveMode.Append).jdbc(url,"table_name",prop)

中文版戳我

Scala를 사용하면 다음 명령을 사용할 수 있습니다.

sudo -u root spark-shell --jars /mnt/resource/lokeshtest/guava-12.0.1.jar,/mnt/resource/lokeshtest/hadoop-aws-2.6.0.jar,/mnt/resource/lokeshtest/aws-java-sdk-1.7.3.jar,/mnt/resource/lokeshtest/mysql-connector-java-5.1.38/mysql-connector-java-5.1.38/mysql-connector-java-5.1.38-bin.jar --packages com.databricks:spark-csv_2.10:1.2.0

import org.apache.spark.sql.SQLContext

val sqlcontext = new org.apache.spark.sql.SQLContext(sc)

val dataframe_mysql = sqlcontext.read.format("jdbc").option("url", "jdbc:mysql://Public_IP:3306/DB_NAME").option("driver", "com.mysql.jdbc.Driver").option("dbtable", "tblage").option("user", "sqluser").option("password", "sqluser").load()

dataframe_mysql.show()

스칼라의 경우.sbt이것 또한 효과가 있을 것입니다.

당신의build.sbt파일:

libraryDependencies ++= Seq(
    "org.apache.spark" %% "spark-core" % "1.6.2",
    "org.apache.spark" %% "spark-sql" % "1.6.2",
    "org.apache.spark" %% "spark-mllib" % "1.6.2",
    "mysql" % "mysql-connector-java" % "5.1.12"
)

그럼 기사님 이용 신고만 해주시면 됩니다.

Class.forName("com.mysql.jdbc.Driver").newInstance

val conf = new SparkConf().setAppName("MY_APP_NAME").setMaster("MASTER")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

val data = sqlContext.read
.format("jdbc")
.option("url", "jdbc:mysql://<HOST>:3306/<database>")
.option("user", <USERNAME>)
.option("password", <PASSWORD>)
.option("dbtable", "MYSQL_QUERY")
.load()

자바(maven 사용)의 경우, pom.xml 파일에 스파크 종속성과 sql 드라이버 종속성을 추가합니다.

<properties>
    <java.version>1.8</java.version>
    <spark.version>1.6.3</spark.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
 <dependencies>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>6.0.6</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.10</artifactId>
        <version>${spark.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>${spark.version}</version>
    </dependency>

    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.11</version>
        <scope>test</scope>
    </dependency>
</dependencies>

샘플 코드, mysql이 로컬에 위치하고 데이터베이스 이름이 test, 사용자 이름이 root, password가 password이고 test db의 두 테이블이 table1과 table2라고 가정합니다.

SparkConf sparkConf = new SparkConf();
SparkContext sc = new SparkContext("local", "spark-mysql-test", sparkConf);
SQLContext sqlContext = new SQLContext(sc);

// here you can run sql query
String sql = "(select * from table1 join table2 on table1.id=table2.table1_id) as test_table";
// or use an existed table directly
// String sql = "table1";
DataFrame dataFrame = sqlContext
    .read()
    .format("jdbc")
    .option("url", "jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true")
    .option("user", "root")
    .option("password", "password")
    .option("dbtable", sql)
    .load();

// continue your logical code
......

public static void main(String[] args) {
    Map<String, String> options = new HashMap<String, String>();
    options.put("url","jdbc:postgresql://<DBURL>:<PORT>/<Database>?user=<UserName>&password=<Password>");
    options.put("dbtable", "<TableName>");
    JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("DBConnection").setMaster("local[*]"));
    SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);
    // DataFrame jdbcDF = sqlContext.load("jdbc", options).cache();
    DataFrame jdbcDF = sqlContext.jdbc(options.get("url"),options.get("dbtable"));
    System.out.println("Data------------------->" + jdbcDF.toJSON().first());
    Row[] rows = jdbcDF.collect();
    System.out.println("Without Filter \n ------------------------------------------------- ");
    for (Row row2 : rows) {
        System.out.println(row2.toString());
    }
    System.out.println("Filter Data\n ------------------------------------------------- ");
    jdbcDF = jdbcDF.select("agency_id","route_id").where(jdbcDF.col("route_id").$less$eq(3));
    rows = jdbcDF.collect();
    for (Row row2 : rows) {
        System.out.println(row2.toString());
    }
}

자바의 경우, 이것이 저에게 효과적이었습니다.

@Bean
public SparkConf sparkConf() {
    SparkConf sparkConf = new SparkConf()
            .setAppName(appName)
            .setSparkHome(sparkHome)
            .setMaster(masterUri);

    return sparkConf;
}

@Bean
public JavaSparkContext javaSparkContext() {
    return new JavaSparkContext(sparkConf());
}

@Bean
public SparkSession sparkSession() {
    return SparkSession
            .builder()
            .sparkContext(javaSparkContext().sc())
            .appName("Java Spark SQL basic example")
            .getOrCreate();
}

Properties properties = new Properties();
        properties.put("user", "root");
        properties.put("password", "root");
        properties.put("driver", "com.mysql.cj.jdbc.Driver");
        sparkSession.read()
                    .jdbc("jdbc:mysql://localhost:3306/books?useSSL=false", "(SELECT books.BOOK_ID as BOOK_ID, books.BOOK_TITLE as BOOK_TITLE, books.BOOK_AUTHOR as BOOK_AUTHOR, borrowers.BORR_NAME as BORR_NAME FROM books LEFT OUTER JOIN borrowers ON books.BOOK_ID = borrowers.BOOK_ID) as t", properties) // join example
                    .show();

물론 MySQL의 경우 커넥터가 필요했습니다.

    <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>6.0.6</version>
    </dependency>

그리고 나는.

+-------+------------------+--------------+---------------+
|BOOK_ID|        BOOK_TITLE|   BOOK_AUTHOR|      BORR_NAME|
+-------+------------------+--------------+---------------+
|      1|        Gyűrű kúra|J.R.K. Tolkien|   Sára Sarolta|
|      2|     Kecske-eledel|     Mekk Elek|Maláta Melchior|
|      3|      Répás tészta| Vegán Eleazár|           null|
|      4|Krumpli és pityóka| Farmer Emília|           null|
+-------+------------------+--------------+---------------+

이 인포오브젝트 기사를 바탕으로 다음을 시도합니다(Java 또는 Scala를 가정할 때 python에서 이것이 어떻게 작동할지 확실하지 않음).

  • mysql-connector-java를 스파크 클러스터 경로에 추가합니다.
  • 드라이버 초기화:Class.forName("com.mysql.jdbc.Driver")
  • JdbcRDD 데이터 원본 생성:
val myRDD = new JdbcRDD( sc, () => 
                               DriverManager.getConnection(url,username,password),
                        "select first_name,last_name,gender from person limit ?, ?",
                        1,//lower bound
                        5,//upper bound
                        2,//number of partitions
                        r =>
                          r.getString("last_name") + ", " + r.getString("first_name"))

   val query: String =
    "select col1, col2 from schema.table_name where condition"

  val url= "jdbc:mysql://<ip>:3306/<schema>"
  val username = ""
  val password = ""
  val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  val df = sqlContext.load("jdbc", Map(
    "url" -> (url + "/?user=" + username + "&password=" + password),
    "dbtable" -> s"($query) as tbl",
    "driver" -> "com.mysql.jdbc.Driver"))

df.show()

Spark 2.1.0 및 Scala(Windows 7 OS 상)의 경우 아래 코드가 꽤 잘 작동합니다.

import org.apache.spark.sql.SparkSession

object MySQL {
  def main(args: Array[String]) {
    //At first create a Spark Session as the entry point of your app
    val spark:SparkSession = SparkSession
      .builder()
      .appName("JDBC")
      .master("local[*]")
      .config("spark.sql.warehouse.dir", "C:/Exp/")
      .getOrCreate();    

    val dataframe_mysql = spark.read.format("jdbc")
                          .option("url", "jdbc:mysql://localhost/feedback")
                          .option("driver", "com.mysql.jdbc.Driver")
                          .option("dbtable", "person") //replace with own
                          .option("user", "root") //replace with own 
                          .option("password", "vertrigo") // replace with own
                          .load()

    dataframe_mysql.show()
  }
}

언급URL : https://stackoverflow.com/questions/27718382/how-to-integrate-apache-spark-with-mysql-for-reading-database-tables-as-a-spark

반응형

'programing' 카테고리의 다른 글

x.jade 부분을 각도로 렌더링하는 것이 가능합니까?  (0) 2023.10.18
로컬 테이블에서 참조 테이블로의 SQL 사용 값이 다른 DB에 있음  (0) 2023.10.18
Excel: 수식을 제거하지만 답은 텍스트로 유지  (0) 2023.10.13
특정 폴더 안의 모든 파일을 읽는 방법  (0) 2023.10.13
AngularJS - 모델 값을 변경하는 지시에서 $render를 호출해야 하는 이유는 무엇입니까?  (0) 2023.10.13

'programing'의 다른글

  • 현재글데이터베이스 테이블을 스파크 데이터 프레임으로 읽기 위해 Apache Spark와 MySQL을 통합하는 방법은 무엇입니까?

관련글

댓글

티스토리툴바