|
#
首先 本文中的 hadoop join 在實際開發沒有用處！
如果在開發中 請使用 cascading groupby, 進行 hadoop join,
本文只是為探討弄懂 cascading 實現做準備。
當然 如果有有人 hadoop join 過 請聯系我，大家交流下 ！
文件可能需要的一些參考：
hadoop jython ( windows )
jython ，jython 編譯以及jar 包
少量 linux shell
本文介紹 hadoop 可能使用到的 join 接口測試 ，已經參考：
使用Hadoop實現Inner Join操作的方法【from淘寶】：http://labs.chinamobile.com/groups/58_547
下面 測試后 ，我這大體上 對 hadoop join 的方式是這樣理解的 （猜想）：
數據1 ; 數據2
job1.map( 數據1 ) =（臨時文件1）> 文件標示1+需要join列 數據
job2.map( 數據2 ) =（臨時文件2）> 文件標示2+需要join列 數據
臨時文件 mapred.join.expr 生成
job3.map ->
文件標示1+需要join列 : 數據
文件標示2+需要join列 : 數據
......
job3.Combiner - >
需要join列 : 文件標示1+數據
需要join列 : 文件標示2+數據
job3.Reducer->
需要join列 : 使用 java-list > 生成
文件2-列x [ 數據,數據... ]
文件1-列x [ 數據,數據... ]
然后 你這 left join ,或 inner join 或 xxx join 邏輯 就自己來吧
結果集合
[root@localhost python]# cat /home/megajobs/del/jobs/tools/hadoop-0.18.3/data/090907/1
1
2
3
4
5
[root@localhost python]# cat /home/megajobs/del/jobs/tools/hadoop-0.18.3/data/090907/2
2
4
3
1
修改 ..../hadoop-0.18.3/src/examples/python/compile
#!/usr/bin/env bash
export HADOOP_HOME=/home/xx/del/jobs/tools/hadoop-0.18.3
export CASCADING_HOME=/home/xx/del/jobs/tools/cascading-1.0.16-hadoop-0.18.3
export JYTHON_HOME=/home/xx/del/jobs/tools/jython2.2.1
export CLASSPATH="$HADOOP_HOME/hadoop-0.18.3-core.jar"
# so that filenames w/ spaces are handled correctly in loops below
IFS=
# add libs to CLASSPATH
for f in $HADOOP_HOME/lib/*.jar; do
CLASSPATH=${CLASSPATH}:$f;
done
for f in $HADOOP_HOME/lib/jetty-ext/*.jar; do
CLASSPATH=${CLASSPATH}:$f;
done
for f in $CASCADING_HOME/*.jar; do
CLASSPATH=${CLASSPATH}:$f;
done
for f in $CASCADING_HOME/lib/*.jar; do
CLASSPATH=${CLASSPATH}:$f;
done
for f in $JYTHON_HOME/*.jar; do
CLASSPATH=${CLASSPATH}:$f;
done
# restore ordinary behaviour
unset IFS
/home/xx/del/jobs/tools/jython2.2.1/jythonc -p org.apache.hadoop.examples -d -j $1.jar -c $1.py
/home/xx/del/jobs/tools/hadoop-0.18.3/bin/hadoop jar $1.jar $2 $3 $4 $5 $6 $7 $8 $9
簡單 數據 鏈接 :
from org.apache.hadoop.fs import Path
from org.apache.hadoop.io import *
from org.apache.hadoop.mapred.lib import *
from org.apache.hadoop.mapred.join import *
from org.apache.hadoop.mapred import *
import sys
import getopt
class tMap(Mapper, MapReduceBase):
def map(self, key, value, output, reporter):
output.collect( Text( str(key) ) , Text( value.toString() ))
def main(args):
conf = JobConf(tMap)
conf.setJobName("wordcount")
conf.setMapperClass( tMap )
FileInputFormat.setInputPaths(conf,[ Path(sp) for sp in args[1:-1]])
conf.setOutputKeyClass( Text )
conf.setOutputValueClass( Text )
conf.setOutputPath(Path(args[-1]))
JobClient.runJob(conf)
if __name__ == "__main__":main(sys.argv)
運行
./compile test file:///home/xx/del/jobs/tools/hadoop-0.18.3/data/090907/1 file:///home/xx/del/jobs/tools/hadoop-0.18.3/data/090907/2 file:///home/xx/del/jobs/tools/hadoop-0.18.3/tmp/wc78
結果:
[xx@localhost wc78]$ cat ../wc78/part-00000
0 1
0 2
2 4
2 2
4 3
4 3
6 1
6 4
8 5
簡單的數據 join :
from org.apache.hadoop.fs import Path
from org.apache.hadoop.io import *
from org.apache.hadoop.mapred.lib import *
from org.apache.hadoop.mapred.join import *
from org.apache.hadoop.mapred import *
import sys
import getopt
class tMap(Mapper, MapReduceBase):
def map(self, key, value, output, reporter):
output.collect( Text( str(key) ) , Text( value.toString() ))
def main(args):
conf = JobConf(tMap)
conf.setJobName("wordcount")
conf.setMapperClass( tMap )
conf.set("mapred.join.expr", CompositeInputFormat.compose("override",TextInputFormat, args[1:-1] ) )
conf.setOutputKeyClass( Text )
conf.setOutputValueClass( Text )
conf.setInputFormat(CompositeInputFormat)
conf.setOutputPath(Path(args[-1]))
JobClient.runJob(conf)
if __name__ == "__main__":main(sys.argv)
運行結果 ( ) :
./compile test file:///home/xx/del/jobs/tools/hadoop-0.18.3/data/090907/1 file:///home/xx/del/jobs/tools/hadoop-0.18.3/data/090907/2 file:///home/xx/del/jobs/tools/hadoop-0.18.3/tmp/wc79
[xx@localhost wc78]$ cat ../wc79/part-00000
0 2
2 4
4 3
6 1
8 5
參考 ： hadoop window 搭建 后,由于對 py 的語法喜歡 ，一直想 把hadoop,改成jython 的
這次 在 自己電腦上 終于 完成,下面介紹過程:
測試環境：
依然的 windows + cygwin
hadoop 0.18 # C:/cygwin/home/lky/tools/java/hadoop-0.18.3
jython 2.2.1 # C:/jython2.2.1
參考: PythonWordCount
啟動 hadoop 并到 hdoop_home 下
# 在云環境中創建 input 目錄
$>bin/hadoop dfs -mkdir input
# 在 包 hadoop 的 NOTICE.txt 拷貝到 input 目錄下
$>bin/hadoop dfs -copyFromLocal c:/cygwin/home/lky/tools/java/hadoop-0.18.3/NOTICE.txt hdfs:///user/lky/input
$>cd src/examples/python
# 創建 個 腳本 ( jy->jar->hd run ) 一步完成!
# 當然 在 linux 寫個腳本比這 好看 呵呵！
$>vim run.bat
"C:\Program Files\Java\jdk1.6.0_11\bin\java.exe" -classpath "C:\jython2.2.1\jython.jar;%CLASSPATH%" org.python.util.jython C:\jython2.2.1\Tools\jythonc\jythonc.py -p org.apache.hadoop.examples -d -j wc.jar -c %1
sh C:\cygwin\home\lky\tools\java\hadoop-0.18.3\bin\hadoop jar wc.jar %2 %3 %4 %5 %6 %7 %8 %9
# 修改 jythonc 打包 環境 。 +hadoop jar
$>vim C:\jython2.2.1\Tools\jythonc\jythonc.py
# Copyright (c) Corporation for National Research Initiatives
# Driver script for jythonc2. See module main.py for details
import sys,os,glob
for fn in glob.glob('c:/cygwin/home/lky/tools/java/hadoop-0.18.3/*.jar') :sys.path.append(fn)
for fn in glob.glob('c:/jython2.2.1/*.jar') :sys.path.append(fn)
for fn in glob.glob('c:/cygwin/home/lky/tools/java/hadoop-0.18.3/lib/*.jar') :sys.path.append(fn)
import main
main.main()
import os
os._exit(0)
# 運行
C:/cygwin/home/lky/tools/java/hadoop-0.18.3/src/examples/python>
run.bat WordCount.py hdfs:///user/lky/input file:///c:/cygwin/home/lky/tools/java/hadoop-0.18.3/tmp2
結果輸出：
cat c:/cygwin/home/lky/tools/java/hadoop-0.18.3/tmp2/part-00000
(http://www.apache.org/). 1
Apache 1
Foundation 1
Software 1
The 1
This 1
by 1
developed 1
includes 1
product 1
software 1
下面重頭來了 ：（簡潔的 jy hdoop 代碼）
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from org.apache.hadoop.fs import Path
from org.apache.hadoop.io import *
from org.apache.hadoop.mapred import *
import sys
import getopt
class WordCountMap(Mapper, MapReduceBase):
one = IntWritable(1)
def map(self, key, value, output, reporter):
for w in value.toString().split():
output.collect(Text(w), self.one)
class Summer(Reducer, MapReduceBase):
def reduce(self, key, values, output, reporter):
sum = 0
while values.hasNext():
sum += values.next().get()
output.collect(key, IntWritable(sum))
def printUsage(code):
print "wordcount [-m <maps>] [-r <reduces>] <input> <output>"
sys.exit(code)
def main(args):
conf = JobConf(WordCountMap);
conf.setJobName("wordcount");
conf.setOutputKeyClass(Text);
conf.setOutputValueClass(IntWritable);
conf.setMapperClass(WordCountMap);
conf.setCombinerClass(Summer);
conf.setReducerClass(Summer);
try:
flags, other_args = getopt.getopt(args[1:], "m:r:")
except getopt.GetoptError:
printUsage(1)
if len(other_args) != 2:
printUsage(1)
for f,v in flags:
if f == "-m":
conf.setNumMapTasks(int(v))
elif f == "-r":
conf.setNumReduceTasks(int(v))
conf.setInputPath(Path(other_args[0]))
conf.setOutputPath(Path(other_args[1]))
JobClient.runJob(conf);
if __name__ == "__main__":
main(sys.argv)
謝謝 同事 孫超 講解
這就 把他的 思想 畫個圖
代碼:
# -*- coding: UTF8 -*-
import sys
# 最小 支持度
sup_min = int(sys.argv[1])
ss = ","
# 交易 數據 庫
D=[
'A,B,C,D',
'B,C,E',
'A,B,C,E',
'B,D,E',
'A,B,C,D'
]
print "交易數據庫展現"
for arr in D : print arr
print
'''
rows=int(sys.argv[1])
D=[]
for tid in open('BuyMusic.20090722.mob.prodIds').readlines()[:rows] :
D.append(tid.split("\n")[0].split("\t")[1])
print "讀取 文件結束 BuyMusic.20090722.mob.prodIds !"
'''
#全局 頻繁項 收集
sup_data_map = {}
#全局 最大頻繁項 收集
is_zsup={}
# 遍歷過程 臨時 局部 頻繁項 收集
mapL = {}
# 第一次 頻繁項 收集
def find_frequent_1_itemset(I):
if I=='null' or I=='' : return
if mapL.has_key(I): mapL[I]+=1
else: mapL[I]=1
map(find_frequent_1_itemset,[ I for TID in D for I in TID.split(ss) ])
# 刷選掉 小于 最小支持度 的 頻繁項
def remove_not_sup_min(map,supmin=sup_min):
for k in [k for k,v in map.items() if v<supmin] :
del map[k]
remove_not_sup_min(mapL)
print "第一次 篩選 頻繁項 結束!"
print mapL
# 裝載 全局 頻繁項 最大頻繁項
for k,v in mapL.items() :
sup_data_map[k]=v
is_zsup[k]=v
# 判定 是否 'BD' 屬于 'BCD' 中
isInTid = lambda I,TID : len(I.split(ss)) == len([i for i in I if i in TID.split(ss)])
# 組合 [A,B] + [A,C] = [A,B.C]
def comb(arr1,arr2):
tmap={}
for v in arr1+arr2 : tmap[v]=""
return tmap.keys()
# apriori 迭代核心
def runL(mapL,dep):
mapL2 = {}
C={}
keys = mapL.keys()
iik=""
jjk=""
# 根據 上次 頻繁項 ，生成本次 '可能頻繁項' 集合
for ii in range(len(keys)) :
for jj in range(ii+1,len(keys)) :
keystr=comb([ch for ch in keys[ii].split(ss)],[ch for ch in keys[jj].split(ss)])
if not len(keystr) == dep : continue
keystr.sort()
tk=ss.join(keystr)
if not tk in C : C[tk]=(keys[ii],keys[jj])
# '可能頻繁項' 對比 交易數據庫 計數
for tk,z in C.items():
for TID in D:
if isInTid(tk,TID) :
if mapL2.has_key(tk): mapL2[tk]+=1
else: mapL2[tk]=1
# 刷選掉 小于 最小支持度 的 頻繁項
remove_not_sup_min(mapL2)
for k,v in is_zsup.items() :
for k1,v1 in mapL2.items() :
if isInTid(k,k1) :
del is_zsup[k]
break
# 全局 頻繁項 ,最大頻繁項 收集
for k,v in mapL2.items() :
sup_data_map[k]=v
is_zsup[k]=v
print "第"+str(dep)+"次 篩選 頻繁項 結束!"
return mapL2
# 真正 運行
ii=1
while mapL :
ii=ii+1
mapL = runL(mapL,ii)
print mapL
# 全局 頻繁項 中 去除 最大頻繁項
for k,v in is_zsup.items() :
if sup_data_map.has_key(k) : del sup_data_map[k]
print "頻繁項"
print sup_data_map
print
print "最大頻繁項"
print is_zsup
print
print "可信度 展現"
for k,v in sup_data_map.items() :
for k1,v1 in is_zsup.items() :
if isInTid(k,k1) :
print k,"->",k1,"\t%.1f" %((float(is_zsup[k1])/float(sup_data_map[k]))*100)+"%"
結果:
-bash-3.00$ python ap.py 2
交易數據庫展現
A,B,C,D
B,C,E
A,B,C,E
B,D,E
A,B,C,D
第一次 篩選 頻繁項 結束!
{'A': 3, 'C': 4, 'B': 5, 'E': 3, 'D': 3}
第2次 篩選 頻繁項 結束!
{'C,D': 2, 'C,E': 2, 'A,D': 2, 'A,B': 3, 'A,C': 3, 'B,E': 3, 'B,D': 3, 'B,C': 4}
第3次 篩選 頻繁項 結束!
{'A,B,D': 2, 'A,B,C': 3, 'B,C,D': 2, 'B,C,E': 2, 'A,C,D': 2}
第4次 篩選 頻繁項 結束!
{'A,B,C,D': 2}
第5次 篩選 頻繁項 結束!
{}
頻繁項
{'A': 3, 'C': 4, 'B': 5, 'E': 3, 'D': 3, 'C,D': 2, 'C,E': 2, 'A,D': 2, 'A,B': 3, 'A,C': 3, 'A,B,D': 2, 'B,C,D': 2, 'A,C,D': 2, 'B,E': 3, 'B,D': 3, 'B,C': 4, 'A,B,C': 3}
最大頻繁項
{'B,C,E': 2, 'A,B,C,D': 2}
可信度 展現
A -> A,B,C,D 66.7%
C -> B,C,E 50.0%
C -> A,B,C,D 50.0%
B -> B,C,E 40.0%
B -> A,B,C,D 40.0%
E -> B,C,E 66.7%
D -> A,B,C,D 66.7%
C,D -> A,B,C,D 100.0%
C,E -> B,C,E 100.0%
A,D -> A,B,C,D 100.0%
A,B -> A,B,C,D 66.7%
A,C -> A,B,C,D 66.7%
A,B,D -> A,B,C,D 100.0%
B,C,D -> A,B,C,D 100.0%
A,C,D -> A,B,C,D 100.0%
B,E -> B,C,E 66.7%
B,D -> A,B,C,D 66.7%
B,C -> B,C,E 50.0%
B,C -> A,B,C,D 50.0%
A,B,C -> A,B,C,D 66.7%
一些特殊正則元字符說明:
1. *? 和 +? 和 {n,}? 懶惰匹配
1.1 非懶惰 ↓
echo "ab2c121a" |perl -ne 'print $1 if /(.*)"d/;' #print ab2c12
1.2 懶惰 ↓
echo "ab2c121a" |perl -ne 'print $1 if /(.*?)"d/;' #print ab
2. 回溯引用和前后查找:
2.1 向前查找 (?=..) ↓
echo "ab2c121a" |perl -ne 'print $1 if /(.*?)(?=2)/;' #print ab
2.2 向后查找 (?<=..) ↓
echo "ab2c121a" |perl -ne 'print $1 if /(?<=2)(.*)(?=2)/;' #print c1
2.3 負向-前/后 查找 (?!..) (?<!..)
#不能匹配 .. ↓
echo "ab2c121a" |perl -ne 'print $1 if /(?<!2)(c.*)/;' #print 無
echo "ab2c121a" |perl -ne 'print $1 if /(?<!3)(c.*)/;' #print c121a
2.4 條件 ?() = if ?()| = if else
# ?() 例如 <p> </p> 必須同時出現 ↓ ↓
echo "<p>xx</p>"|perl -ne 'print $2 if /(<p>)?("w*)(?(1)<"/p>)/' #print xx
echo "<p>xx"|perl -ne 'print $2,""n" if /(<p>)?("w*)(?(1)<"/p>)/' #print 空
echo "xx"|perl -ne 'print $2 if /(<p>)?("w*)(?(1)<"/p>)/' #print xx
# ?()| 例如 還是上面的，
# 當 有<p> 可以接</p> 也可以接 數字結尾 ↓
echo "<p>xx1</p>"|perl -ne 'print $2 if /(<p>)?("w*)(?(1)<"/p>|"d)/' #print xx1
echo "<p>xx1"|perl -ne 'print $2 if /(<p>)?("w*)(?(1)<"/p>|"d)/' # print xx
當熟悉 hash db python bsddb （db-key 轉）
使用確實很方便，但是沒有 想 關系數據庫中的 select order by 查詢 ，感覺比較郁悶! 上網 一頓 google ......
import bsddb
db = bsddb.btopen('/tmp/spam.db', 'c')
for i in range(10): db['%d'%i] = '%d'% (i*i)
db['3'] # 9
db.keys() # ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
db.set_location('6') # 36
db.previous() # 25
db.next() # 36
db.next() # 47
這可以定位，并且 previous , next 什么的 （不過目前好像是針對 string 自然 排序！）
這里比較實用的 demo
import bsddb
db = bsddb.btopen('/tmp/spam2.db', 'c')
db["2009-08-14 22:00"]="gg"
db["2009-08-15 22:00"]="cc"
db["2009-07-15 00:00"]="tt"
db["2009-08-16 22:00"]="gg"
# 注意 這 統配 等價 正則 = 2009-08-15.*
# 開始 以為能使用 正則 ，但不能 。只能簡單的 xxx.* 形式的
db.set_location('2009-08-15') # ('2009-08-15 22:00', 'cc')
db.next() # ('2009-08-16 22:00', 'gg')
db.set_location('2009-08-15') # ('2009-08-15 22:00', 'cc')
db.previous() #('2009-08-14 22:00', 'gg')
轉:http://www.daniweb.com/forums/thread31449.html
什么都不說了，直接看代碼吧。
注解 應該寫的比較詳細
# liukaiyi
# 注 k-means ，維度類型 - 數值形式 ( 199 或 23.13  )
import sys, math, random
# -- 類化 '數據'
# 在 n-維度空間
class Point:
def __init__(self, coords, reference=None):
self.coords = coords
self.n = len(coords)
self.reference = reference
def __repr__(self):
return str(self.coords)
# -- 類化 '聚集點 / 聚類平均距離 點 '
# -- 在 n-維度空間
# -- k-means 核心類
# -- 每次 聚集各點 圍繞她 進行聚集
# -- 并提供方法 求-聚集后的計算中心點，同時記入 此次 中心點(聚集各點平均距離)，為下一次聚集提供中心點.
class Cluster:
def __init__(self, points):
if len(points) == 0: raise Exception("ILLEGAL: EMPTY CLUSTER")
self.points = points
self.n = points[0].n
for p in points:
if p.n != self.n: raise Exception("ILLEGAL: MULTISPACE CLUSTER")
# 求 聚集各點后 平均點
self.centroid = self.calculateCentroid()
def __repr__(self):
return str(self.points)
# 更新 中心點，并返回 原中心點 與 現中心點(聚集各點平均距離)距離
def update(self, points):
old_centroid = self.centroid
self.points = points
self.centroid = self.calculateCentroid()
return getDistance(old_centroid, self.centroid)
# 計算平均點 （聚集/收集各點（離本類的中心點）最近數據,后生成新的 中心點 ）
def calculateCentroid(self):
centroid_coords = []
# 維度 迭代
for i in range(self.n):
centroid_coords.append(0.0)
# 收集各點 迭代
for p in self.points:
centroid_coords[i] = centroid_coords[i]+p.coords[i]
centroid_coords[i] = centroid_coords[i]/len(self.points)
return Point(centroid_coords)
# -- 返回根據 k-means 聚集形成的 數據集
def kmeans(points, k, cutoff):
# Randomly sample k Points from the points list, build Clusters around them
initial = random.sample(points, k)
clusters = []
for p in initial: clusters.append(Cluster([p]))
# 迭代 k-means 直到 每次迭代 各收集點 別的 最多 不超過 0.5
while True:
# k 個收集 數組
lists = []
for c in clusters: lists.append([])
# 迭代 每個 數據點 ，并計算與每個中心點距離
# 并把數據點添加入相應最短的中心點收集數組中
# 在迭代中 smallest_distance 為每個點與各中心點最短距離 參數，請注意看
for p in points:
smallest_distance = getDistance(p, clusters[0].centroid)
index = 0
for i in range(len(clusters[1:])):
distance = getDistance(p, clusters[i+1].centroid)
if distance < smallest_distance:
smallest_distance = distance
index = i+1
# 添加到 離最短中心距離的 數組中
lists[index].append(p)
# 聚集完，計算新 中心點
# 并 cluster.centroid 屬性記入下 新中心點（下一次 聚集的中心點 ）
# 并 計算與上一次 中心點 距離 ，如果 差值在 cutoff 0.5 以下 ,跳出迭代 （結束，返回最后一次 聚集集合）
biggest_shift = 0.0
for i in range(len(clusters)):
shift = clusters[i].update(lists[i])
biggest_shift = max(biggest_shift, shift)
if biggest_shift < cutoff: break
return clusters
# -- 得到歐幾里德距離兩點之間
def getDistance(a, b):
# Forbid measurements between Points in different spaces
if a.n != b.n: raise Exception("ILLEGAL: NON-COMPARABLE POINTS")
# Euclidean distance between a and b is sqrt(sum((a[i]-b[i])^2) for all i)
ret = 0.0
for i in range(a.n):
ret = ret+pow((a.coords[i]-b.coords[i]), 2)
return math.sqrt(ret)
# -- 在 n-維度 空間中創建 隨機點
# -- 隨機生成 測試數據
def makeRandomPoint(n, lower, upper):
coords = []
for i in range(n): coords.append(random.uniform(lower, upper))
return Point(coords)
# main
def main(args):
# 參數說明
# num_points, n, k, cutoff, lower, upper
# 隨機數據數量 , 維度, 聚集數, 跳出迭代最小距離 , 維度數最大值,維度數最小值
num_points, n, k, cutoff, lower, upper = 10, 2, 3, 0.5, -200, 200
# 在 n-維度空間里 , 創建 num_points 隨機點
# 測試數據生成
points = []
for i in range(num_points): points.append(makeRandomPoint(n, lower, upper))
# 使用 k-means 算法，來 聚集數據點 (算法入口點)
clusters = kmeans(points, k, cutoff)
print "\nPOINTS:"
for p in points: print "P:", p
print "\nCLUSTERS:"
for c in clusters: print "C:", c
if __name__ == "__main__": main(sys.argv)
1. vi /etc/vsftpd/vsftpd.conf
添加：
listen=YES
tcp_wrappers=YES
port_enable=YES
ftp_data_port=20
listen_port=21
listen_address=0.0.0.0
port_promiscuous=NO
no_anon_password=NO
anon_mkdir_write_enable=no
2.將chroot_list_enable=YES前的#去掉
并將chroot_list_file=/etc/vsftpd.chroot_list 前的#去掉
3.創建用戶
useradd 用戶
passwd 用戶
4. vi /etc/vsftpd.chroot_list
將 用戶 添加到文件里
5.修改用戶的登錄路徑（主目錄）
vi /etc/passwd
如：data:x:516:516::/home/data/data:/sbin/nologin
6.啟動vsftp
service vsftpd restart
Java 代碼：
package com.xunjie.dmsp.olduser;
import java.util.Properties;
import cascading.flow.Flow;
import cascading.flow.FlowConnector;
import cascading.operation.regex.RegexSplitter;
import cascading.pipe.Each;
import cascading.pipe.Pipe;
import cascading.scheme.TextLine;
import cascading.tap.Hfs;
import cascading.tap.Tap;
import cascading.tuple.Fields;
/**
* test.txt：
* 1 a
* 2 b
* 3 c
*
* /data/hadoop/hadoop/bin/hadoop jar
* dmsp_test_jar-1.0-SNAPSHOT-dependencies.jar
* hdfs:/user/hadoop/test/lky/test.txt
* file:///data/hadoop/test/lky/output
*/
public class Test2 {
public static void main(String[] args) {
//設定輸入文件
String sourcePath= args[0];
//設置輸出文件夾
String sinkPath = args[1];
//定義讀取列
Fields inputfields = new Fields("num", "value");
//定義分解正則，默認 \t
RegexSplitter spliter = new RegexSplitter(inputfields);
//管道定義
Pipe p1 = new Pipe( "test" );
//管道嵌套：
//分解日志源文件，輸出給定字段
p1 = new Each(p1,new Fields("line") ,spliter);
//設定輸入和輸出 ，使用 泛型Hfs
Tap source = new Hfs( new TextLine(), sourcePath );
Tap sink = new Hfs( new TextLine() , sinkPath );
//配置job
Properties properties = new Properties();
properties.setProperty("hadoop.job.ugi", "hadoop,hadoop");
FlowConnector.setApplicationJarClass( properties, Main.class );
FlowConnector flowConnector = new FlowConnector(properties);
Flow importFlow = flowConnector.connect( "import flow", source,sink,p1);
importFlow.start();
importFlow.complete();
}
}
這特殊關注下， 開啟慢查詢。在web開發中很有幫助
MYSQL啟用日志，和查看日志
|
| 時間：2009-01-21 17:33:57 來源：http://wasabi.javaeye.com/blog/318962 作者：kenbli |
mysql有以下幾種日志：
錯誤日志： -log-err
查詢日志： -log
慢查詢日志: -log-slow-queries
更新日志: -log-update
二進制日志： -log-bin
是否啟用了日志
mysql>show variables like 'log_%';
怎樣知道當前的日志
mysql> show master status;
顯示二進制日志數目
mysql> show master logs;
看二進制日志文件用mysqlbinlog
shell>mysqlbinlog mail-bin.000001
或者shell>mysqlbinlog mail-bin.000001 | tail
在配置文件中指定log的輸出位置.
Windows：Windows 的配置文件為 my.ini，一般在 MySQL 的安裝目錄下或者 c:\Windows 下。
Linux：Linux 的配置文件為 my.cnf ，一般在 /etc 下。
在linux下：
- # 在[mysqld] 中輸入
- #log
- log-error=/usr/local/mysql/log/error.log
- log=/usr/local/mysql/log/mysql.log
- long_query_time=2
- log-slow-queries= /usr/local/mysql/log/slowquery.log
windows下:
- # 在[mysqld] 中輸入
- #log
- log-error="E:/PROGRA~1/EASYPH~1.0B1/mysql/logs/error.log"
- log="E:/PROGRA~1/EASYPH~1.0B1/mysql/logs/mysql.log"
- long_query_time=2
- log-slow-queries= "E:/PROGRA~1/EASYPH~1.0B1/mysql/logs/slowquery.log"
開啟慢查詢
long_query_time =2 --是指執行超過多久的sql會被log下來，這里是2秒
log-slow-queries= /usr/local/mysql/log/slowquery.log --將查詢返回較慢的語句進行記錄
log-queries-not-using-indexes = nouseindex.log --就是字面意思，log下來沒有使用索引的query
log=mylog.log --對所有執行語句進行記錄 |
|