Hadoop集羣（第10期）_MySQL關係數據庫

時間 2019-11-09

標籤 hadoop 集羣 mysql 關係數據庫欄目 Hadoop 简体版

原文原文鏈接

一、MySQL安裝

MySQL下載地址：http://www.mysql.com/downloads/java

1.1 Windows平臺

　　1）準備軟件 node

　　MySQL版本：mysql-5.5.21-win32.msimysql

　　2）安裝環境： linux

　　操做系統：Windows 7旗艦版web

　　3）開始安裝 sql

　　第一步：雙擊"msi"安裝文件，出現如圖1.1-1界面——"MySQL安裝嚮導"，按"Next"繼續。數據庫

圖1.1-1 MySQL安裝嚮導apache

　　第二步：在"I accept …."前面勾上，贊成協議，按"Next"按鈕繼續。編程

圖1.1-2 軟件協議vim

　　第三步：選擇安裝類型，有"Typical（默認）"、"Custom（定製安裝）"、"Complete（徹底）"三個選項。

典型安裝：安裝只安裝MySQL服務器、mysql命令行客戶端和命令行實用程序。命令行客戶端和實用程序包括mysqldump、myisamchk和其它幾個工具來幫助你管理MySQL服務器。
定製安裝：安裝容許你徹底控制你想要安裝的軟件包和安裝路徑。
徹底安裝：安裝將安裝軟件包內包含的全部組件。徹底安裝軟件包包括的組件包括嵌入式服務器庫、基準套件、支持腳本和文檔。

　　咱們選擇"Custom"，有更多的選項，也方便熟悉安裝過程。

圖1.1-3 安裝類型

　　第四步：選擇組件及更改文件夾位置。

圖1.1-4 自定義界面

　　全部可用組件列入定製安裝對話框左側的樹狀視圖內。未安裝的組件用紅色X 圖標表示；已經安裝的組件有灰色圖標。要想更改組件，點擊該組件的圖標並從下拉列表中選擇新的選項。組件我選擇了默認安裝，位置我會更改一下，點擊Browse。

圖1.1-5 路徑選擇

　　按"OK"按鈕返回，並按"Next"按鈕繼續。

　　備註：安裝mysql的路徑中，不能含有中文。

　　第五步：確認一下先前的設置，若是有誤，按"Back"返回重作。按"Install"開始安裝。

圖1.1-6 準備安裝

　　第六步：正在安裝中，請稍候……

圖1.1-7 正在安裝

　　第七步：彈出一個頁面來，是關於介紹MySQL企業版的信息，沒有什麼可操做的，按"Next"按鈕繼續。

圖1.1-8 MySQL企業版介紹

　　而後彈出一個相似界面，接着按"Next"按鈕繼續。

　　第八步：那個帶複選框的是"MySQL服務器實例配置嚮導"，保持默認勾選，按"Finish"按鈕。至此，安裝過程已經結束了，可是還須要配置一下。

圖1.1-9 安裝結束

　　第九步：MySQL配置嚮導啓動界面，按"Next"按鈕繼續。

圖1.1-10 配置嚮導

　　MySQL Configuration Wizard（配置嚮導）能夠幫助自動配置Windows中的服務器。MySQL Configuration Wizard（配置嚮導）問你一系列問題，而後將回答放到模板中生成一個my.ini文件，該文件與你的安裝一致。目前只適用於Windows用戶。

　　通常狀況當MySQL安裝幫助退出時，從MySQL安裝幫助啓動MySQL Configuration Wizard（配置嚮導）。還能夠點擊Windows啓動菜單中MySQL服務器實例配置嚮導條目中的MySQL部分來啓動MySQL Configuration Wizard（配置嚮導）。而且，還能夠進入MySQL安裝bin目錄直接啓動MySQLInstanceConfig.exe文件。

　　第十步：選擇配置類型，能夠選擇兩種配置類型：Detailed Configuration（詳細配置）和Standard Configuration（標準配置）。Standard Configuration（標準配置）選項適合想要快速啓動MySQL而沒必要考慮服務器配置的新用戶。詳細配置選項適合想要更加細粒度控制服務器配置的高級用戶。咱們選擇"Detailed Configuration"，方便熟悉配置過程。

圖1.1-11 配置類型

　　備註：

　　若是你是MySQL的新手，須要配置爲單用戶開發機的服務器，Standard Configuration（標準配置）應當適合你的需求。選擇Standard Configuration（標準配置）選項，則 MySQL Configuration Wizard（配置嚮導）自動設置全部配置選項，但不包括服務選項和安全選項。

　　Standard Configuration（標準配置）設置選項可能與安裝MySQL的系統不兼容。若是系統上已經安裝了MySQL和你想要配置的安裝，建議選擇詳細配置。

　　第十一步：選擇服務器類型，能夠選擇3種服務器類型，選擇哪一種服務器將影響到MySQL Configuration Wizard（配置嚮導）對內存、硬盤和過程或使用的決策。

Developer Machine（開發機器）：該選項表明典型我的用桌面工做站。假定機器上運行着多個桌面應用程序。將MySQL服務器配置成使用最少的系統資源。

Server Machine（服務器）：該選項表明服務器，MySQL服務器能夠同其它應用程序一塊兒運行，例如FTP、email和web服務器。MySQL服務器配置成使用適當比例的系統資源。

Dedicated MySQL Server Machine（專用MySQL服務器）：該選項表明只運行MySQL服務的服務器。假定運行沒有運行其它應用程序。MySQL服務器配置成使用全部可用系統資源。

你們根據本身的類型選擇了，通常選"Server Machine"，不會太少，也不會佔滿。咱們選擇"Server Machine"，按"Next"按鈕繼續。

圖1.1-12 服務器類型

　　第十二步：選擇數據庫用途，經過Database Usage（數據庫使用）對話框，你能夠指出建立MySQL表時使用的表處理器。經過該選項，你能夠選擇是否使用InnoDB儲存引擎，以及InnoDB佔用多大比例的服務器資源。"Multifunctional Database（通用多功能型，好）"、"Transactional Database Only（服務器類型，專一於事務處理，通常）"、"Non-Transactional Database Only（非事務處理型，較簡單，主要作一些監控、記數用，對MyISAM 數據類型的支持僅限於non-transactional）"，隨本身的用途而選擇了，通常選擇第一種多功能的。咱們選擇"Multifunctional Database"，按"Next"按鈕繼續。

圖1.1-13 數據庫用途

　　第十三步：對InnoDB Tablespace 進行配置，就是爲InnoDB 數據庫文件選擇一個存儲空間，若是修改了，要記住位置，重裝的時候要選擇同樣的地方，不然可能會形成數據庫損壞，固然，對數據庫作個備份就沒問題了，這裏不詳述。這裏沒有修改，使用用默認位置，按"Next"按鈕繼續。

圖1.1-14 配置InnoDB表空間

　　第十四步：選擇MySQL容許的最大鏈接數，限制所建立的與MySQL服務器之間的並行鏈接數量很重要，以便防止服務器耗盡資源。在Concurrent Connections（並行鏈接）對話框中，能夠選擇服務器的使用方法，並根據狀況限制並行鏈接的數量。還能夠手動設置並行鏈接的限制。第一種是最大20個鏈接併發數，第二種是最大500個併發鏈接數，最後一種是自定義。咱們選擇"Online Transaction PRocessing(OLTP)"，按"Next"按鈕繼續。

圖1.1-15 最大鏈接數

　　第十五步：進行網絡配置，在Networking Options（網絡選項）對話框中能夠啓用或禁用TCP/IP網絡，並配置用來鏈接MySQL服務器的端口號。默認狀況啓用TCP/IP網絡。要想禁用TCP/IP網絡，取消選擇Enable TCP/IP Networking選項旁邊的檢查框。默認使用3306端口。要想更訪問MySQL使用的端口，從下拉框選擇一個新端口號或直接向下拉框輸入新的端口號。若是你選擇的端口號已經被佔用，將提示確認選擇的端口號。咱們保持默認選項，按"Next"按鈕繼續。

圖1.1-16 網絡配置

　　第十六步：對數據庫語言編碼進行設置，很是重要，由於Hadoop裏默認編碼爲UTF-8，因此爲了不出現亂碼，咱們這裏選擇"UTF-8"做爲MySQL數據庫的語言編碼。

圖1.1-17 數據庫編碼

　　第十七步：是否要把MySQL設置成Windows的服務，通常選擇設成服務，這樣之後就能夠經過服務中啓動和關閉mysql數據庫了。推薦：下面的複選框也勾選上，這樣，在cmd模式下，沒必要非到mysql的bin目錄下執行命令。咱們所有打上了勾，Service Name 不變。按"Next"按鈕繼續。

圖1.1-18 服務選項

　　第十八步：設置MySQL的超級用戶密碼，這個超級用戶很是重要，對MySQL擁有所有的權限，請設置好並牢記超級用戶的密碼，下面有個複選框是選擇是否容許遠程機器用root用戶鏈接到你的MySQL服務器上面，若是有這個需求，也請勾選。咱們這裏的root用戶密碼設置爲"hadoop"，並勾上"容許遠程鏈接"複選框，按"Next"按鈕繼續。

圖1.1-19 安全選項

　　備註：

"Enable root access from remote machines（是否容許root 用戶在其它的機器上登錄，若是要安全，就不要勾上，若是要方便，就勾上它）"。
"Create An Anonymous Account（新建一個匿名用戶，匿名用戶能夠鏈接數據庫，不能操做數據，包括查詢）"，通常就不用勾了。

　　第十九步：確認設置無誤，若是有誤，按"Back"返回檢查。若是沒有，按"Execute"使設置生效。

圖1.1-20 確認配置

　　第二十步：設置完畢，按"Finish"按鈕結束MySQL的安裝與配置。

圖1.1-21 配置完成

　　備註：這裏有一個比較常見的錯誤，就是不能"Start service"，通常出如今之前有安裝MySQL的服務器上，解決的辦法，先保證之前安裝的MySQL 服務器完全卸載掉了；不行的話，檢查是否按上面一步所說，以前的密碼是否有修改，照上面的操做；若是依然不行，將MySQL 安裝目錄下的data 文件夾備份，而後刪除，在安裝完成後，將安裝生成的 data 文件夾刪除，備份的data 文件夾移回來，再重啓MySQL 服務就能夠了，這種狀況下，可能須要將數據庫檢查一下，而後修復一次，防止數據出錯。

　　4）驗證成功

　　第一種：打開任務管理器看到MySQL服務是否已經啓動。

圖1.1-22 任務管理器

　　第二種："開始à啓動cmdà開打cmd模式"，輸入"mysql –u root –p"鏈接數據庫。

圖1.1-23 鏈接數據庫

1.2 Linux平臺

　　1）準備軟件

　　MySQL數據庫：MySQL-server-5.5.21-1.linux2.6.i386.rpm

　　MySQL客戶端：MySQL-client-5.5.21-1.linux2.6.i386.rpm

　　2）安裝環境：

　　操做系統：CentOS6.0 Linux

　　3）檢查安裝

　　在安裝MySQL以前，先檢查CentOS系統中是否已經安裝了一個MySQL，若是已經安裝先卸載，否則會致使安裝新的MySQL失敗。

　　用下面命令查看系統以前是否已安裝MySQL。

rpm -qa | grep mysql

　　查看結果以下：

　　從上圖得知，CentOS6.0系統自帶了一個MySQL，咱們須要刪除這個老版本，用root用戶執行下面語句。

rpm -e --nodeps mysql-libs-5.1.47-4.el6.i686

　　上圖中，咱們先切換到"root"用戶下，而後執行刪除語句，刪除以後，咱們再次查看，發現已經成功刪除了CentOS6.0自帶的舊MySQL版本。

　　在刪除MySQL的rpm後，還要進行一些掃尾操做，網上有兩種操做。（備註：我在這裏兩種都沒有用到，發現系統中並無其餘殘餘的MySQL信息。）

　　第一種善後處理：使用下面命令進行處理。

rm -rf /var/lib/mysql*

rm -rf /usr/share/mysql*

　　另外一種善後處理：卸載後/var/lib/mysql中的/etc/my.cnf會重命名爲my.cnf.rpmsave，/var/log/mysqld.log 會重命名爲/var/log/mysqld.log.rpmsave，若是肯定沒用後就手工刪除。

　　4）開始安裝

　　第一步：上傳所需軟件。經過"FlashFXP"軟件使用"vsftpd"上傳用到的兩個軟件到"/home/hadoop"目錄下。

　　第二步：安裝MySQL服務端。用"root"用戶運行以下命令進行安裝：（備註：如下步驟都是用"root"用戶執行。）

rpm -ivh MySQL-server-5.5.21-1.linux2.6.i386.rpm

　　經過SecureCRT查看以下：

　　如出現如上信息，服務端安裝完畢。

　　第三步：檢測MySQL 3306端口是否安打開。測試是否成功可運行netstat看MySQL端口是否打開，如打開表示服務已經啓動，安裝成功。MySQL默認的端口是3306。

netstat -nat

　　從上圖中發現並無與"3306"有關的信息，說明"MySQL服務器"沒有啓動。經過下面命令啓動MySQL。

service mysql start

　　從上圖中已經發現咱們的MySQL服務器已經起來了。

　　第四步：安裝MySQL客戶端。用下面命令進行安裝：

rpm -ivh MySQL-client-5.5.21-1.linux2.6.i386.rpm

　　執行命令顯示以下：

　　從上圖中顯示MySQL客戶端已經安裝完畢。

　　第五步：MySQL的幾個重要目錄。MySQL安裝完成後不像SQL Server默認安裝在一個目錄，它的數據庫文件、配置文件和命令文件分別在不一樣的目錄，瞭解這些目錄很是重要，尤爲對於Linux的初學者，由於 Linux自己的目錄結構就比較複雜，若是搞不清楚MySQL的安裝目錄那就無從談起深刻學習。

　　下面就介紹一下這幾個目錄。

a、數據庫目錄

/var/lib/mysql/

b、配置文件

/usr/share/mysql（mysql.server命令及配置文件）

c、相關命令

/usr/bin(mysqladmin mysqldump等命令)

d、啓動腳本

/etc/rc.d/init.d/（啓動腳本文件mysql的目錄）

如：/etc/rc.d/init.d/mysql start/restart/stop/status

　　下面就分別展現上面的幾個目錄內容：

配置文件

相關命令

啓動腳本

　　第六步：更改MySQL目錄。因爲MySQL數據庫目錄佔用磁盤比較大，而MySQL默認的數據文件存儲目錄爲/"var/lib/mysql"，因此咱們要把目錄移到"/"根目錄下的"mysql_data"目錄中。

　　須要如下幾個步驟：

"/"根目錄下創建"mysql_data"目錄

cd /

mkdir mysql_data

把MySQL服務進程停掉

　　能夠用兩種方法：

service mysql stop

　　或者

mysqladmin -u root -p shutdown

　　從上圖中咱們得知"MySQL服務進程"已經停掉。

　　備註：MySQL默認用戶名爲"root"，此處的"root"與Linux的最高權限用戶"root"不是一下子，並且默認的用戶"root"的密碼爲空，因此上圖中讓輸入密碼，直接點擊回車便可。

把"/var/lib/mysql"整個目錄移到"/mysql_data"

mv /var/lib/mysql /mysql_data

　　這樣就把MySQL的數據文件移動到了"/mysql_data/mysql"下。

找到my.cnf配置文件

　　若是"/etc/"目錄下沒有my.cnf配置文件，請到"/usr/share/mysql/"下找到*.cnf文件，拷貝其中一個合適的配置文件到"/etc/"並更名爲"my.cnf"中。命令以下：

cp /usr/share/mysql/my-medium.cnf　 /etc/my.cnf

　　上圖中，下查看"/etc/"下面是否有"my.cnf"文件，發現沒有，而後經過上面的命令進行拷貝，拷貝完以後，進行查看，發現拷貝成功。

　　備註："/usr/share/mysql/"下有好幾個結尾爲cnf的文件，它們的做用分別是。

a、my-small.cnf：是爲了小型數據庫而設計的。不該該把這個模型用於含有一些經常使用項目的數據庫。

b、my-medium.cnf：是爲中等規模的數據庫而設計的。若是你正在企業中使用RHEL,可能會比這個操做系統的最小RAM需求(256MB)明顯多得多的物理內存。因而可知，若是有那麼多RAM內存可使用，天然能夠在同一臺機器上運行其它服務。

c、my-large.cnf：是爲專用於一個SQL數據庫的計算機而設計的。因爲它能夠爲該數據庫使用多達512MB的內存，因此在這種類型的系統上將須要至少1GB的RAM,以便它可以同時處理操做系統與數據庫應用程序。

d、my-huge.cnf：是爲企業中的數據庫而設計的。這樣的數據庫要求專用服務器和1GB或1GB以上的RAM。

這些選擇高度依賴於內存的數量、計算機的運算速度、數據庫的細節大小、訪問數據庫的用戶數量以及在數據庫中裝入並訪問數據的用戶數量。隨着數據庫和用戶的不斷增長，數據庫的性能可能會發生變化。

　　備註：這裏咱們根據實際狀況，選擇了"my-medium.cnf"進行配置。

編輯MySQL的配置文件"/etc/my.cnf"

　　爲保證MySQL可以正常工做，須要指明"mysql.sock"文件的產生位置，以及默認編碼修改成UTF-8。用下面命令：

vim /etc /my.cnf

　　須要修改和添加的內容以下：

【client】

socket = /mysql_data/mysql/mysql.sock

default-character-set=utf8

【mysqld】

socket = /mysql_data/mysql/mysql.sock

datadir =/mysql_data/mysql

character-set-server=utf8

lower_case_table_names=1（注意linux下mysql安裝完後是默認：區分表名的大小寫，不區分列名的大小寫；lower_case_table_names = 0 0：區分大小寫，1：不區分大小寫）

　　備註：【client】和【mysqld】設置的編碼時前地名稱不同。

修改MySQL啓動腳本"/etc/rc.d/init.d/mysql"

　　最後，須要修改MySQL啓動腳本/etc/rc.d/init.d/mysql，修改datadir=/mysql_data/mysql。

vim /etc/rc.d/init.d/mysql

從新啓動MySQL服務

service mysql start

　　正準備高興時，發現MySQL啓動不了了，網上搜了一下午，各類都沒有解決。後來在一篇文章才得知又是"SELinux"惹得禍。解決辦法以下：

　　打開/etc/selinux/config，把SELINUX=enforcing改成SELINUX=disabled後存盤退出重啓機器試試，必需要重啓，很關鍵。

　　機器重啓以後，在把"mysql服務"啓動。

　　第七步：修改登陸密碼。

　　MySQL默認沒有密碼，安裝完畢增長密碼的重要性是不言而喻的。

修改前，直接登陸

　　在沒有添加密碼前，直接輸入"mysql"就能登陸到MySQL數據庫裏。

修改登陸密碼

　　用到的命令以下：

mysqladmin -u root password 'new-password'

格式：mysqladmin -u用戶名 -p舊密碼 password 新密碼

　　咱們這裏設置MySQL數據庫"root"用戶的密碼爲"hadoop"。執行的命令以下：

mysqladmin –u root password hadoop

測試是否修改爲功

　　（1）不用密碼登陸

　　此時顯示錯誤，說明密碼已經修改。

　　（2）用修改後的密碼登陸

　　從上圖中得知，咱們已經成功修改了密碼，而且用新的密碼登陸了MySQL服務器。

　　第八步：配置防火牆

　　第一種：修改防火牆配置文件"/etc/sysconfig/iptables"，添加以下內容：

-A INPUT -m state --state NEW -m tcp -p tcp --sport 3306 -j ACCEPT

-A OUTPUT -m state --state NEW -m tcp -p tcp --dport 3306 -j ACCEPT

　　而後執行下面命令，使防火牆當即生效。

service iptables restart

　　第二種：關閉防火牆

　　經過下面兩個命令使防火牆關閉，而且永遠不起做用。

service iptables stop

chkconfig iptables off

　　咱們在這裏爲了方便，採用第二種方法，執行效果以下。

第九步：驗證MySQL數據庫編碼是否爲UTF-8。

　　鏈接上數據庫以後，輸入命令："SHOW VARIABLES LIKE '%char%';"便可查看到如今你的數據庫所使用的字符集了。

　　第十步：刪除空用戶，加強安全。

　　目前爲止咱們都是以"root"的身份進行的，可是當咱們切換至普通用戶登陸MySQL時，直接輸入"mysql"就進去了，咱們剛纔不是設置密碼了嗎？怎麼就失效了呢？說明有空用戶存在。先用命令"exit"退出，在按照下面命令進行修正。

解決步驟以下：

以MySQL用戶"root"用密碼形式登陸。

mysql -u root -p

刪除空用戶，強烈建議。

mysql>delete from mysql.user where user='';

刷新權限表，以即可以使更改當即生效。

mysql>flush privileges;

輸入"exit"，退出MySQL。

mysql>exit

再從新以"mysql"登陸測試

mysql

發現以"mysql"登陸已經失效，必須以"mysql –u root -p"才能登陸。

下面是執行效果截圖：

二、MapReduce與MySQL交互

　　MapReduce技術推出後，曾遭到關係數據庫研究者的挑剔和批評，認爲MapReduce不具有有相似於關係數據庫中的結構化數據存儲和處理能力。爲此，Google和MapReduce社區進行了不少努力。一方面，他們設計了相似於關係數據中結構化數據表的技術（Google的BigTable，Hadoop的HBase）提供一些粗粒度的結構化數據存儲和處理能力；另外一方面，爲了加強與關係數據庫的集成能力，Hadoop MapReduce提供了相應的訪問關係數據庫庫的編程接口。

　　MapReduce與MySQL交互的總體架構以下圖所示。

圖2-1整個環境的架構

　　具體到MapReduce框架讀/寫數據庫，有2個主要的程序分別是 DBInputFormat和DBOutputFormat，DBInputFormat 對應的是SQL語句select，而DBOutputFormat 對應的是 Inster/update，使用DBInputFormat和DBOutputForma時候須要實現InputFormat這個抽象類，這個抽象類含有getSplits()和createRecordReader()抽象方法，在DBInputFormat類中由 protected String getCountQuery() 方法傳入結果集的個數，getSplits()方法再肯定輸入的切分原則，利用SQL中的 LIMIT 和 OFFSET 進行切分得到數據集的範圍，請參考DBInputFormat源碼中public InputSplit[] getSplits(JobConf job, int chunks) throws IOException的方法，在DBInputFormat源碼中createRecordReader()則能夠按必定格式讀取相應數據。

1）創建關係數據庫鏈接

DBConfiguration：提供數據庫配置和建立鏈接的接口。

DBConfiguration類中提供了一個靜態方法建立數據庫鏈接：

public static void configureDB(Job job,String driverClass,String dbUrl,String userName,String Password)

其中，job爲當前準備執行的做業，driverClasss爲數據庫廠商提供的訪問其數據庫的驅動程序，dbUrl爲運行數據庫的主機的地址，userName和password分別爲數據庫提供訪問地用戶名和相應的訪問密碼。

2）相應的從關係數據庫查詢和讀取數據的接口

DBInputFormat：提供從數據庫讀取數據的格式。
DBRecordReader：提供讀取數據記錄的接口。

　　3）相應的向關係數據庫直接輸出結果的編程接口

DBOutputFormat：提供向數據庫輸出數據的格式。
DBRecordWrite：提供數據庫寫入數據記錄的接口。

　　數據庫鏈接完成後，便可完成從MapReduce程序向關係數據庫寫入數據的操做。爲了告知數據庫將寫入哪一個表中的哪些字段，DBOutputFormat中提供了一個靜態方法來指定須要寫入的數據表和字段：

public static void setOutput(Job job,String tableName,String ... fieldName)

其中，tableName指定即將寫入的數據表，後續參數將指定哪些字段數據將寫入該表。

2.1 從數據庫中輸入數據

雖然Hadoop容許從數據庫中直接讀取數據記錄做爲MapReduce的輸入，但處理效率較低，並且大量頻繁地從MapReduce程序中查詢和讀取關係數據庫可能會大大增長數據庫的訪問負載，所以DBInputFormat僅適合讀取小量數據記錄的計算和應用，不適合數據倉庫聯機數據分析大量數據的讀取處理。

讀取大量數據記錄一個更好的解決辦法是：用數據庫中的Dump工具將大量待分析數據輸出爲文本數據文件，並上載到HDFS中進行處理。

1）首先建立要讀入的數據

Windows環境

　　首先建立數據庫"school"，使用下面命令進行：

create database school;

而後經過如下幾句話，把咱們事先準備好的sql語句（student.sql事先放到了D盤目錄）導入到剛建立的"school"數據庫中。用到的命令以下：

use school;

source d:\student.sql

"student.sql"中的內容以下所示：

DROP TABLE IF EXISTS `school`.`student`;

CREATE TABLE `school`.`student` (

`id` int(11) NOT NULL default '0',

`name` varchar(20) default NULL,

`sex` varchar(10) default NULL,

`age` int(10) default NULL,

PRIMARY KEY (`id`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

INSERT INTO `student` VALUES ('201201', '張三', '男', '21');

INSERT INTO `student` VALUES ('201202', '李四', '男', '22');

INSERT INTO `student` VALUES ('201203', '王五', '女', '20');

INSERT INTO `student` VALUES ('201204', '趙六', '男', '21');

INSERT INTO `student` VALUES ('201205', '小紅', '女', '19');

INSERT INTO `student` VALUES ('201206', '小明', '男', '22');

執行結果以下所示：

查詢剛纔建立的數據庫表"student"的內容。

結果發現顯示是亂碼，記得我當時是設置的UTF-8，怎麼就出現亂碼了呢？其實咱們使用的操做系統的系統爲中文，且它的默認編碼是gbk，而MySQL的編碼有兩種，它們分別是：

　　【client】：客戶端的字符集。客戶端默認字符集。當客戶端向服務器發送請求時，請求以該字符集進行編碼。

　　【mysqld】：服務器字符集，默認狀況下所採用的。

找到安裝MySQL目錄，好比咱們的安裝目錄爲：

E:\HadoopWorkPlat\MySQL Server 5.5

從中找到"my.ini"配置文件，最終發現my.ini裏的2個character_set把client改爲gbk，把server改爲utf8就能夠了。

【client】端：

[client]

port=3306

[mysql]

default-character-set=gbk

【mysqld】端：

[mysqld]

# The default character set that will be used when a new schema or table is

# created and no character set is defined

character-set-server=utf8

按照上面修改完以後，重啓MySQL服務。

此時在Windows下面的數據庫表已經準備完成了。

Linux環境

　　首先經過"FlashFXP"把咱們剛纔的"student.sql"上傳到"/home/hadoop"目錄下面，而後按照上面的語句建立"school"數據庫。

查看咱們上傳的"student.sql"內容：

建立"school"數據庫，並導入"student.sql"語句。

顯示數據庫"school"中的表"student"信息。

顯示錶"student"中的內容。

到此爲止在"Windows"和"Linux"兩種環境下面都建立了表"student"表，並初始化了值。下面就開始經過MapReduce讀取MySQL庫中表"student"的信息。

2）使MySQL能遠程鏈接

MySQL默認是容許別的機器進行遠程訪問地，爲了使Hadoop集羣能訪問MySQL數據庫，因此進行下面操做。

用MySQL用戶"root"登陸。

mysql -u root -p

使用下面語句進行受權，賦予任何主機訪問數據的權限。

GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY 'hadoop' WITH GRANT OPTION;

刷新，使之當即生效。

FLUSH PRIVILEGES;

執行結果以下圖。

Windows下面：

Linux下面：

到目前爲止，若是鏈接Win7上面的MySQL數據庫還不行，你們還應該記得前面在Linux下面關掉了防火牆，可是咱們在Win7下對防火牆並無作任何處理，若是不對防火牆作處理，即便執行了上面的遠程受權，仍然不能鏈接。下面是設置Win7上面的防火牆，使遠程機器能經過3306端口訪問MySQL數據庫。

解決方案：只要在'入站規則'上創建一個3306端口便可。

　　執行順序：控制面板à管理工具à高級安全的Windows防火牆à入站規則

　　而後新建規則à選擇'端口'à在'特定本地端口'上輸入一個'3306' à選擇'容許鏈接'=>選擇'域'、'專用'、'公用'=>給個名稱，如：MySqlInput

3）對JDBC的Jar包處理

由於程序雖然用Eclipse編譯運行但最終要提交到Hadoop集羣上，因此JDBC的jar必須放到Hadoop集羣中。有兩種方式：

（1）在每一個節點下的${HADOOP_HOME}/lib下添加該包，重啓集羣，通常是比較原始的方法。

咱們的Hadoop安裝包在"/usr/hadoop"，因此把Jar放到"/usr/hadoop/lib"下面，而後重啓，記得是Hadoop集羣中全部的節點都要放，由於執行分佈式是程序是在每一個節點本地機器上進行。

（2）在Hadoop集羣的分佈式文件系統中建立"/lib"文件夾，並把咱們的的JDBC的jar包上傳上去，而後在主程序添加以下語句，就能保證Hadoop集羣中全部的節點都能使用這個jar包。由於這個jar包放在了HDFS上，而不是本地系統，這個要理解清楚。

DistributedCache.addFileToClassPath(new Path("/lib/mysql-connector-java-5.1.18-bin.jar"), conf);

咱們用的JDBC的jar以下所示：

mysql-connector-java-5.1.18-bin.jar

經過Eclipse下面的DFS Locations進行建立"/lib"文件夾，並上傳JDBC的jar包。執行結果以下：

備註：咱們這裏採用了第二種方式。

4）源程序代碼以下所示

package com.hebut.mr;

import java.io.IOException;

import java.io.DataInput;

import java.io.DataOutput;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import org.apache.hadoop.filecache.DistributedCache;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.Writable;

import org.apache.hadoop.mapred.JobClient;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.FileOutputFormat;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.mapred.lib.IdentityReducer;

import org.apache.hadoop.mapred.lib.db.DBWritable;

import org.apache.hadoop.mapred.lib.db.DBInputFormat;

import org.apache.hadoop.mapred.lib.db.DBConfiguration;

public class ReadDB {

    public static class Map extends MapReduceBase implements

            Mapper<LongWritable, StudentRecord, LongWritable, Text> {

        // 實現map函數

        public void map(LongWritable key, StudentRecord value,

        OutputCollector<LongWritable, Text> collector, Reporter reporter)

                throws IOException {

            collector.collect(new LongWritable(value.id),

                    new Text(value.toString()));

        }

    }

    public static class StudentRecord implements Writable, DBWritable {

        public int id;

        public String name;

        public String sex;

        public int age;

        @Override

        public void readFields(DataInput in) throws IOException {

            this.id = in.readInt();

            this.name = Text.readString(in);

            this.sex = Text.readString(in);

            this.age = in.readInt();

        }

        @Override

        public void write(DataOutput out) throws IOException {

            out.writeInt(this.id);

            Text.writeString(out, this.name);

            Text.writeString(out, this.sex);

            out.writeInt(this.age);

        }

        @Override

        public void readFields(ResultSet result) throws SQLException {

            this.id = result.getInt(1);

            this.name = result.getString(2);

            this.sex = result.getString(3);

            this.age = result.getInt(4);

        }

        @Override

        public void write(PreparedStatement stmt) throws SQLException {

            stmt.setInt(1, this.id);

            stmt.setString(2, this.name);

            stmt.setString(3, this.sex);

            stmt.setInt(4, this.age);

        }

        @Override

        public String toString() {

            return new String("學號：" + this.id + "_姓名：" + this.name

                    + "_性別："+ this.sex + "_年齡：" + this.age);

        }

    }

    public static void main(String[] args) throws Exception {

        JobConf conf = new JobConf(ReadDB.class);

        // 這句話很關鍵

        conf.set("mapred.job.tracker", "192.168.1.2:9001");

        // 很是重要，值得關注

        DistributedCache.addFileToClassPath(new Path(

         "/lib/mysql-connector-java-5.1.18-bin.jar"), conf);

        // 設置輸入類型

        conf.setInputFormat(DBInputFormat.class);

        // 設置輸出類型

        conf.setOutputKeyClass(LongWritable.class);

        conf.setOutputValueClass(Text.class);

        // 設置Map和Reduce類

        conf.setMapperClass(Map.class);

        conf.setReducerClass(IdentityReducer.class);

        // 設置輸出目錄

        FileOutputFormat.setOutputPath(conf, new Path("rdb_out"));

        // 創建數據庫鏈接

        DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver",

            "jdbc:mysql://192.168.1.24:3306/school", "root", "hadoop");

        // 讀取"student"表中的數據

        String[] fields = { "id", "name", "sex", "age" };

        DBInputFormat.setInput(conf, StudentRecord.class, "student", null,"id", fields);

        JobClient.runJob(conf);

    }

}

備註：因爲Hadoop1.0.0新的API對關係型數據庫暫不支持，只能用舊的API進行，因此下面的"向數據庫中輸出數據"也是如此。

5）運行結果以下所示

通過上面的設置後，已經經過鏈接Win7和Linux上的MySQL數據庫，執行結果都同樣。惟獨變得就是代碼中"DBConfiguration.configureDB"中MySQL數據庫所在機器的IP地址。

2.2 向數據庫中輸出數據

基於數據倉庫的數據分析和挖掘輸出結果的數據量通常不會太大，於是可能適合於直接向數據庫寫入。咱們這裏嘗試與"WordCount"程序相結合，把單詞統計的結果存入到關係型數據庫中。

1）建立寫入的數據庫表

咱們還使用剛纔建立的數據庫"school"，只是在裏添加一個新的表"wordcount"，仍是使用下面語句執行：

use school;

source sql腳本全路徑

下面是要建立的"wordcount"表的sql腳本。

DROP TABLE IF EXISTS `school`.`wordcount`;

CREATE TABLE `school`.`wordcount` (

`id` int(11) NOT NULL auto_increment,

`word` varchar(20) default NULL,

`number` int(11) default NULL,

PRIMARY KEY (`id`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

執行效果以下所示：

Windows環境

Linux環境

2）程序源代碼以下所示

package com.hebut.mr;

import java.io.IOException;

import java.io.DataInput;

import java.io.DataOutput;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.util.Iterator;

import java.util.StringTokenizer;

import org.apache.hadoop.filecache.DistributedCache;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.io.Writable;

import org.apache.hadoop.mapred.FileInputFormat;

import org.apache.hadoop.mapred.JobClient;

import org.apache.hadoop.mapred.JobConf;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reducer;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.mapred.TextInputFormat;

import org.apache.hadoop.mapred.lib.db.DBOutputFormat;

import org.apache.hadoop.mapred.lib.db.DBWritable;

import org.apache.hadoop.mapred.lib.db.DBConfiguration;

public class WriteDB {

    // Map處理過程

    public static class Map extends MapReduceBase implements

            Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);

        private Text word = new Text();

        @Override

        public void map(Object key, Text value,

            OutputCollector<Text, IntWritable> output, Reporter reporter)

                throws IOException {

            String line = value.toString();

            StringTokenizer tokenizer = new StringTokenizer(line);

            while (tokenizer.hasMoreTokens()) {

                word.set(tokenizer.nextToken());

                output.collect(word, one);

            }

        }

    }

    // Combine處理過程

    public static class Combine extends MapReduceBase implements

            Reducer<Text, IntWritable, Text, IntWritable> {

        @Override

        public void reduce(Text key, Iterator<IntWritable> values,

            OutputCollector<Text, IntWritable> output, Reporter reporter)

                throws IOException {

            int sum = 0;

            while (values.hasNext()) {

                sum += values.next().get();

            }

            output.collect(key, new IntWritable(sum));

        }

    }

    // Reduce處理過程

    public static class Reduce extends MapReduceBase implements

            Reducer<Text, IntWritable, WordRecord, Text> {

        @Override

        public void reduce(Text key, Iterator<IntWritable> values,

            OutputCollector<WordRecord, Text> collector, Reporter reporter)

                throws IOException {

            int sum = 0;

            while (values.hasNext()) {

                sum += values.next().get();

            }

            WordRecord wordcount = new WordRecord();

            wordcount.word = key.toString();

            wordcount.number = sum;

            collector.collect(wordcount, new Text());

        }

    }

    public static class WordRecord implements Writable, DBWritable {

        public String word;

        public int number;

        @Override

        public void readFields(DataInput in) throws IOException {

            this.word = Text.readString(in);

            this.number = in.readInt();

        }

        @Override

        public void write(DataOutput out) throws IOException {

            Text.writeString(out, this.word);

            out.writeInt(this.number);

        }

        @Override

        public void readFields(ResultSet result) throws SQLException {

            this.word = result.getString(1);

            this.number = result.getInt(2);

        }

        @Override

        public void write(PreparedStatement stmt) throws SQLException {

            stmt.setString(1, this.word);

            stmt.setInt(2, this.number);

        }

    }

    public static void main(String[] args) throws Exception {

        JobConf conf = new JobConf(WriteDB.class);

        // 這句話很關鍵

        conf.set("mapred.job.tracker", "192.168.1.2:9001");

        DistributedCache.addFileToClassPath(new Path(

                "/lib/mysql-connector-java-5.1.18-bin.jar"), conf);

        // 設置輸入輸出類型

        conf.setInputFormat(TextInputFormat.class);

        conf.setOutputFormat(DBOutputFormat.class);

        // 不加這兩句，通不過，可是網上給的例子沒有這兩句。

        conf.setOutputKeyClass(Text.class);

        conf.setOutputValueClass(IntWritable.class);

        // 設置Map和Reduce類

        conf.setMapperClass(Map.class);

        conf.setCombinerClass(Combine.class);

        conf.setReducerClass(Reduce.class);

        // 設置輸如目錄

        FileInputFormat.setInputPaths(conf, new Path("wdb_in"));

        // 創建數據庫鏈接

        DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver",

            "jdbc:mysql://192.168.1.24:3306/school", "root", "hadoop");

        // 寫入"wordcount"表中的數據

        String[] fields = { "word", "number" };

        DBOutputFormat.setOutput(conf, "wordcount", fields);

        JobClient.runJob(conf);

    }

}

3）運行結果以下所示

Windows環境

　　測試數據：

（1）file1.txt

hello word

hello hadoop

（2）file2.txt

蝦皮 hadoop

蝦皮 word

軟件軟件

運行結果：

咱們發現上圖中出現了"?"，後來查找原來是由於個人測試數據時在Windows用記事本寫的而後保存爲"UTF-8"，在保存時爲了區分編碼，自動在前面加了一個"BOM"，可是不會顯示任何結果。然而咱們的代碼把它識別爲"?"進行處理。這就出現了上面的結果，若是咱們在每一個要處理的文件前面的第一行加一個空格，結果就成以下顯示：

接着又作了一個測試，在Linux上面用下面命令建立了一個文件，並寫上中文內容。結果顯示並無出現"?"，並且網上說不一樣的記事本軟件（EmEditor、UE）保存爲"UTF-8"就沒有這個問題。通過修改以後的Map類，就可以正常識別了。

    // Map處理過程

    public static class Map extends MapReduceBase implements

            Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);

        private Text word = new Text();

        @Override

        public void map(Object key, Text value,

            OutputCollector<Text, IntWritable> output, Reporter reporter)

                throws IOException {

            String line = value.toString();



            //處理記事本UTF-8的BOM問題

            if (line.getBytes().length > 0) {

                if ((int) line.charAt(0) == 65279) {

                    line = line.substring(1);

                }

            }



            StringTokenizer tokenizer = new StringTokenizer(line);

            while (tokenizer.hasMoreTokens()) {

                word.set(tokenizer.nextToken());

                output.collect(word, one);

            }

        }

    }

處理以後的結果：

從上圖中得知，咱們的問題已經解決了，所以，在編輯、更改任何文本文件時，請務必使用不會亂加BOM的編輯器。Linux下的編輯器應該都沒有這個問題。Windows下，請勿使用記事本等編輯器。推薦的編輯器是： Editplus 2.12版本以上； EmEditor； UltraEdit（須要取消'添加BOM'的相關選項）； Dreamweaver（須要取消'添加BOM'的相關選項）等。

　　對於已經添加了BOM的文件，要取消的話，能夠用以上編輯器另存一次。（Editplus須要先另存爲gb，再另存爲UTF-8。） DW解決辦法以下：用DW打開指定文件，按Ctrl+Jà標題/編碼à編碼選擇"UTF-8"，去掉"包括Unicode簽名(BOM)"勾選à保存/另存爲，便可。

國外有一個牛人已經把這個問題解決了，使用"UnicodeInputStream"、"UnicodeReader"。

地址：http://koti.mbnet.fi/akini/java/unicodereader/

示例：Java讀帶有BOM的UTF-8文件亂碼緣由及解決方法

代碼：http://download.csdn.net/detail/xia520pi/4146123